1 /***********************************************************************
2  * Copyright (c) 2013-2025 General Atomics Integrated Intelligence, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * https://www.apache.org/licenses/LICENSE-2.0
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.features.avro.io
10 
11 import org.apache.avro.file.{DataFileStream, DataFileWriter}
12 import org.geotools.api.feature.simple.{SimpleFeature, SimpleFeatureType}
13 import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypes
14 
15 object AvroDataFile extends AvroDataFile
16 
17 /**
18   * AvroDataFiles are binary Avro files (see https://avro.apache.org/) that encode
19   * SimpleFeatures using a custom avro schema per SimpleFeatureType. AvroDataFiles
20   * are meant to:
21   * 1. Provide binary longer-term storage in filesystems for SimpleFeatures
22   * 2. Carry the SimpleFeatureType and feature name along with the data
23   *    using avro metadata
24   * 3. Be self-describing outside of Geotools as much as possible
25   *
26   * You may want to consider gzipping your avro data file for better compression
27   *
28   * Version 3 supports Bytes as a type in the SFT
29   */
30 trait AvroDataFile {
31 
32   val SftNameKey = "sft.name"
33   val SftSpecKey = "sft.spec"
34   val VersionKey = "version"
35 
36   private[avro] val Version: Long = 3L
37 
38   def setMetaData(dfw: DataFileWriter[SimpleFeature], sft: SimpleFeatureType): Unit = {
39     dfw.setMeta(VersionKey, Version)
40     dfw.setMeta(SftNameKey, sft.getTypeName)
41     dfw.setMeta(SftSpecKey, SimpleFeatureTypes.encodeType(sft))
42   }
43 
44   /**
45     * Backwards compatible...Version 2 can parse v1
46     *
47     * @param dfs data file stream
48     * @return
49     */
50   def canParse(dfs: DataFileStream[_]): Boolean = {
51     dfs.getMetaKeys.contains(VersionKey) &&
52       dfs.getMetaLong(VersionKey) <= Version &&
53       dfs.getMetaString(SftNameKey) != null &&
54       dfs.getMetaString(SftSpecKey) != null
55   }
56 
57   def getSft(dfs: DataFileStream[_]): SimpleFeatureType = {
58     val sftName = dfs.getMetaString(SftNameKey)
59     val sftString = dfs.getMetaString(SftSpecKey)
60     SimpleFeatureTypes.createType(sftName, sftString)
61   }
62 }