From 51eb07676996a059be5bc0067cf75069c7351d6b Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Sun, 13 Sep 2026 00:11:33 -0700 Subject: [PATCH] Move shared GeoPackage reader and tests into Spark common --- ...pache.spark.sql.sources.DataSourceRegister | 1 + .../geopackage/GeoPackageDataSource.scala | 0 .../GeoPackagePartitionReader.scala | 0 .../GeoPackagePartitionReaderFactory.scala | 0 .../geopackage/GeoPackageScan.scala | 0 .../geopackage/GeoPackageScanBuilder.scala | 0 .../geopackage/GeoPackageTable.scala | 0 .../sedona/sql/GeoPackageReaderTest.scala | 0 ...pache.spark.sql.sources.DataSourceRegister | 1 - ...pache.spark.sql.sources.DataSourceRegister | 1 - .../geopackage/GeoPackageDataSource.scala | 73 --- .../GeoPackagePartitionReader.scala | 107 ---- .../GeoPackagePartitionReaderFactory.scala | 139 ----- .../geopackage/GeoPackageScan.scala | 68 --- .../geopackage/GeoPackageScanBuilder.scala | 71 --- .../geopackage/GeoPackageTable.scala | 119 ----- .../sedona/sql/GeoPackageReaderTest.scala | 476 ------------------ ...pache.spark.sql.sources.DataSourceRegister | 1 - .../geopackage/GeoPackageDataSource.scala | 73 --- .../GeoPackagePartitionReader.scala | 107 ---- .../GeoPackagePartitionReaderFactory.scala | 139 ----- .../geopackage/GeoPackageScan.scala | 68 --- .../geopackage/GeoPackageScanBuilder.scala | 71 --- .../geopackage/GeoPackageTable.scala | 119 ----- .../sedona/sql/GeoPackageReaderTest.scala | 476 ------------------ 25 files changed, 1 insertion(+), 2109 deletions(-) rename spark/{spark-3.5 => common}/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala (100%) rename spark/{spark-3.5 => common}/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala (100%) rename spark/{spark-3.5 => common}/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala (100%) rename spark/{spark-3.5 => common}/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala (100%) rename spark/{spark-3.5 => common}/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala (100%) rename spark/{spark-3.5 => common}/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala (100%) rename spark/{spark-3.5 => common}/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala (100%) delete mode 100644 spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala delete mode 100644 spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala delete mode 100644 spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala delete mode 100644 spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala delete mode 100644 spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala delete mode 100644 spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala delete mode 100644 spark/spark-4.0/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala delete mode 100644 spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala delete mode 100644 spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala delete mode 100644 spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala delete mode 100644 spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala delete mode 100644 spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala delete mode 100644 spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala delete mode 100644 spark/spark-4.1/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala diff --git a/spark/common/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister b/spark/common/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister index 6839972c4a7..13e9d20c556 100644 --- a/spark/common/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister +++ b/spark/common/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister @@ -6,3 +6,4 @@ org.apache.sedona.sql.datasources.osm.OsmPbfFormat org.apache.spark.sql.execution.datasources.geoparquet.GeoParquetFileFormat org.apache.spark.sql.sedona_sql.io.geotiffmetadata.GeoTiffMetadataDataSource org.apache.spark.sql.sedona_sql.io.netcdfmetadata.NetCdfMetadataDataSource +org.apache.sedona.sql.datasources.geopackage.GeoPackageDataSource diff --git a/spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala b/spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala similarity index 100% rename from spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala rename to spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala diff --git a/spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala b/spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala similarity index 100% rename from spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala rename to spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala diff --git a/spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala b/spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala similarity index 100% rename from spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala rename to spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala diff --git a/spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala b/spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala similarity index 100% rename from spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala rename to spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala diff --git a/spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala b/spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala similarity index 100% rename from spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala rename to spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala diff --git a/spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala b/spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala similarity index 100% rename from spark/spark-3.5/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala rename to spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala diff --git a/spark/spark-3.5/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala b/spark/common/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala similarity index 100% rename from spark/spark-3.5/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala rename to spark/common/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala diff --git a/spark/spark-3.5/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister b/spark/spark-3.5/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister index ae1de3d8bd2..c9a3961320e 100644 --- a/spark/spark-3.5/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister +++ b/spark/spark-3.5/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister @@ -1,3 +1,2 @@ org.apache.spark.sql.execution.datasources.v2.geoparquet.metadata.GeoParquetMetadataDataSource org.apache.sedona.sql.datasources.shapefile.ShapefileDataSource -org.apache.sedona.sql.datasources.geopackage.GeoPackageDataSource diff --git a/spark/spark-4.0/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister b/spark/spark-4.0/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister index ae1de3d8bd2..c9a3961320e 100644 --- a/spark/spark-4.0/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister +++ b/spark/spark-4.0/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister @@ -1,3 +1,2 @@ org.apache.spark.sql.execution.datasources.v2.geoparquet.metadata.GeoParquetMetadataDataSource org.apache.sedona.sql.datasources.shapefile.ShapefileDataSource -org.apache.sedona.sql.datasources.geopackage.GeoPackageDataSource diff --git a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala b/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala deleted file mode 100644 index 11f2db38e84..00000000000 --- a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala +++ /dev/null @@ -1,73 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.hadoop.fs.Path -import org.apache.sedona.sql.datasources.geopackage.model.GeoPackageOptions -import org.apache.spark.sql.connector.catalog.Table -import org.apache.spark.sql.execution.datasources.FileFormat -import org.apache.spark.sql.execution.datasources.v2.FileDataSourceV2 -import org.apache.spark.sql.sources.DataSourceRegister -import org.apache.spark.sql.util.CaseInsensitiveStringMap - -import java.util.Locale -import scala.jdk.CollectionConverters._ -import scala.util.Try - -class GeoPackageDataSource extends FileDataSourceV2 with DataSourceRegister { - - override def fallbackFileFormat: Class[_ <: FileFormat] = { - null - } - - override protected def getTable(options: CaseInsensitiveStringMap): Table = { - GeoPackageTable( - "", - sparkSession, - options, - getPaths(options), - None, - fallbackFileFormat, - getLoadOptions(options)) - } - - private def getLoadOptions(options: CaseInsensitiveStringMap): GeoPackageOptions = { - val path = options.get("path") - if (path.isEmpty) { - throw new IllegalArgumentException("GeoPackage path is not specified") - } - - val showMetadata = options.getBoolean("showMetadata", false) - val maybeTableName = options.get("tableName") - - if (!showMetadata && maybeTableName == null) { - throw new IllegalArgumentException("Table name is not specified") - } - - val tableName = if (showMetadata) { - "gpkg_contents" - } else { - maybeTableName - } - - GeoPackageOptions(tableName = tableName, showMetadata = showMetadata) - } - - override def shortName(): String = "geopackage" -} diff --git a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala b/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala deleted file mode 100644 index 4e59163922d..00000000000 --- a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala +++ /dev/null @@ -1,107 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.hadoop.fs.Path -import org.apache.sedona.sql.datasources.geopackage.connection.{FileSystemUtils, GeoPackageConnectionManager} -import org.apache.sedona.sql.datasources.geopackage.model.TableType.{FEATURES, METADATA, TILES, UNKNOWN} -import org.apache.sedona.sql.datasources.geopackage.model.{GeoPackageReadOptions, PartitionOptions, TileRowMetadata} -import org.apache.sedona.sql.datasources.geopackage.transform.ValuesMapper -import org.apache.spark.broadcast.Broadcast -import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.connector.read.PartitionReader -import org.apache.spark.util.SerializableConfiguration - -import java.io.File -import java.sql.ResultSet - -case class GeoPackagePartitionReader( - var rs: ResultSet, - options: GeoPackageReadOptions, - broadcastedConf: Broadcast[SerializableConfiguration], - var currentTempFile: File, - copying: Boolean = false) - extends PartitionReader[InternalRow] { - - private var values: Seq[Any] = Seq.empty - private var currentFile = options.currentFile - private val partitionedFiles = options.partitionedFiles - - override def next(): Boolean = { - if (rs.next()) { - values = ValuesMapper.mapValues(adjustPartitionOptions, rs) - return true - } - - partitionedFiles.remove(currentFile) - - if (partitionedFiles.isEmpty) { - return false - } - - rs.close() - - currentFile = partitionedFiles.head - val (tempFile, _) = FileSystemUtils.copyToLocal( - options = broadcastedConf.value.value, - file = new Path(currentFile.filePath.toString())) - - if (copying) { - currentTempFile.deleteOnExit() - } - - currentTempFile = tempFile - - rs = GeoPackageConnectionManager.getTableCursor(currentTempFile.getPath, options.tableName) - - if (!rs.next()) { - return false - } - - values = ValuesMapper.mapValues(adjustPartitionOptions, rs) - - true - } - - private def adjustPartitionOptions: PartitionOptions = { - options.partitionOptions.tableType match { - case FEATURES | METADATA => options.partitionOptions - case TILES => - val tileRowMetadata = TileRowMetadata( - zoomLevel = rs.getInt("zoom_level"), - tileColumn = rs.getInt("tile_column"), - tileRow = rs.getInt("tile_row")) - - options.partitionOptions.withTileRowMetadata(tileRowMetadata) - case UNKNOWN => options.partitionOptions - } - - } - - override def get(): InternalRow = { - InternalRow.fromSeq(values) - } - - override def close(): Unit = { - rs.close() - if (copying) { - options.tempFile.delete() - } - } -} diff --git a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala b/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala deleted file mode 100644 index 0f2e9a87b81..00000000000 --- a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala +++ /dev/null @@ -1,139 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.hadoop.fs.Path -import org.apache.sedona.sql.datasources.geopackage.connection.{FileSystemUtils, GeoPackageConnectionManager} -import org.apache.sedona.sql.datasources.geopackage.model.TableType.TILES -import org.apache.sedona.sql.datasources.geopackage.model.{GeoPackageOptions, GeoPackageReadOptions, PartitionOptions, TableType} -import org.apache.spark.broadcast.Broadcast -import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.catalyst.expressions.{BoundReference, JoinedRow} -import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection -import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader, PartitionReaderFactory} -import org.apache.spark.sql.execution.datasources.FilePartition -import org.apache.spark.sql.types.StructType -import org.apache.spark.unsafe.types.UTF8String -import org.apache.spark.util.SerializableConfiguration - -case class GeoPackagePartitionReaderFactory( - sparkSession: SparkSession, - broadcastedConf: Broadcast[SerializableConfiguration], - loadOptions: GeoPackageOptions, - dataSchema: StructType, - metadataSchema: StructType) - extends PartitionReaderFactory { - - override def createReader(partition: InputPartition): PartitionReader[InternalRow] = { - val partitionFiles = partition match { - case filePartition: FilePartition => filePartition.files - case _ => - throw new IllegalArgumentException( - s"Unexpected partition type: ${partition.getClass.getCanonicalName}") - } - - val (tempFile, copied) = FileSystemUtils.copyToLocal( - options = broadcastedConf.value.value, - file = new Path(partitionFiles.head.filePath.toString())) - - val tableType = if (loadOptions.showMetadata) { - TableType.METADATA - } else { - GeoPackageConnectionManager.findFeatureMetadata(tempFile.getPath, loadOptions.tableName) - } - - val rs = - GeoPackageConnectionManager.getTableCursor(tempFile.getAbsolutePath, loadOptions.tableName) - - val schema = GeoPackageConnectionManager.getSchema(tempFile.getPath, loadOptions.tableName) - - if (StructType(schema.map(_.toStructField(tableType))) != dataSchema) { - throw new IllegalArgumentException( - s"Schema mismatch: expected $dataSchema, got ${StructType(schema.map(_.toStructField(tableType)))}") - } - - val tileMetadata = tableType match { - case TILES => - Some( - GeoPackageConnectionManager.findTilesMetadata(tempFile.getPath, loadOptions.tableName)) - case _ => None - } - - val baseReader = GeoPackagePartitionReader( - rs = rs, - options = GeoPackageReadOptions( - tableName = loadOptions.tableName, - tempFile = tempFile, - partitionOptions = - PartitionOptions(tableType = tableType, columns = schema, tile = tileMetadata), - partitionedFiles = scala.collection.mutable.HashSet(partitionFiles: _*), - currentFile = partitionFiles.head), - broadcastedConf = broadcastedConf, - currentTempFile = tempFile, - copying = copied) - - if (metadataSchema.nonEmpty) { - val gpkgFile = partitionFiles.head - val filePath = gpkgFile.filePath.toString - val fileName = new Path(filePath).getName - - val allMetadataValues: Map[String, Any] = Map( - "file_path" -> UTF8String.fromString(filePath), - "file_name" -> UTF8String.fromString(fileName), - "file_size" -> gpkgFile.fileSize, - "file_block_start" -> gpkgFile.start, - "file_block_length" -> gpkgFile.length, - "file_modification_time" -> (gpkgFile.modificationTime * 1000L)) - - val innerStructType = metadataSchema.fields.head.dataType.asInstanceOf[StructType] - val prunedValues = innerStructType.fields.map(f => allMetadataValues(f.name)) - val metadataStruct = InternalRow.fromSeq(prunedValues.toSeq) - val metadataRow = InternalRow.fromSeq(Seq(metadataStruct)) - - new PartitionReaderWithMetadata(baseReader, dataSchema, metadataSchema, metadataRow) - } else { - baseReader - } - } -} - -private[geopackage] class PartitionReaderWithMetadata( - reader: PartitionReader[InternalRow], - baseSchema: StructType, - metadataSchema: StructType, - metadataValues: InternalRow) - extends PartitionReader[InternalRow] { - - private val joinedRow = new JoinedRow() - private val unsafeProjection = - GenerateUnsafeProjection.generate(baseSchema.fields.zipWithIndex.map { case (f, i) => - BoundReference(i, f.dataType, f.nullable) - } ++ metadataSchema.fields.zipWithIndex.map { case (f, i) => - BoundReference(baseSchema.length + i, f.dataType, f.nullable) - }) - - override def next(): Boolean = reader.next() - - override def get(): InternalRow = { - unsafeProjection(joinedRow(reader.get(), metadataValues)) - } - - override def close(): Unit = reader.close() -} diff --git a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala b/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala deleted file mode 100644 index edca3d35ff1..00000000000 --- a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala +++ /dev/null @@ -1,68 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.sedona.sql.datasources.geopackage.model.GeoPackageOptions -import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.catalyst.expressions.Expression -import org.apache.spark.sql.connector.read.PartitionReaderFactory -import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex -import org.apache.spark.sql.execution.datasources.v2.FileScan -import org.apache.spark.sql.types.StructType -import org.apache.spark.sql.util.CaseInsensitiveStringMap -import org.apache.spark.util.SerializableConfiguration - -import scala.jdk.CollectionConverters._ - -case class GeoPackageScan( - dataSchema: StructType, - sparkSession: SparkSession, - fileIndex: PartitioningAwareFileIndex, - readDataSchema: StructType, - readPartitionSchema: StructType, - metadataSchema: StructType, - options: CaseInsensitiveStringMap, - loadOptions: GeoPackageOptions) - extends FileScan { - - override def readSchema(): StructType = - StructType(readDataSchema.fields ++ readPartitionSchema.fields ++ metadataSchema.fields) - - override def partitionFilters: Seq[Expression] = { - Seq.empty - } - - override def dataFilters: Seq[Expression] = { - Seq.empty - } - - override def createReaderFactory(): PartitionReaderFactory = { - val caseSensitiveMap = options.asScala.toMap - val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(caseSensitiveMap) - val broadcastedConf = - sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf)) - - GeoPackagePartitionReaderFactory( - sparkSession, - broadcastedConf, - loadOptions, - dataSchema, - metadataSchema) - } -} diff --git a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala b/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala deleted file mode 100644 index 7fdb716f29f..00000000000 --- a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala +++ /dev/null @@ -1,71 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.sedona.sql.datasources.geopackage.model.GeoPackageOptions -import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.connector.read.Scan -import org.apache.spark.sql.execution.datasources.{InMemoryFileIndex, PartitioningAwareFileIndex} -import org.apache.spark.sql.execution.datasources.v2.FileScanBuilder -import org.apache.spark.sql.types.StructType -import org.apache.spark.sql.util.CaseInsensitiveStringMap -import scala.jdk.CollectionConverters._ - -class GeoPackageScanBuilder( - sparkSession: SparkSession, - fileIndex: PartitioningAwareFileIndex, - dataSchema: StructType, - options: CaseInsensitiveStringMap, - loadOptions: GeoPackageOptions, - userDefinedSchema: Option[StructType] = None) - extends FileScanBuilder(sparkSession, fileIndex, dataSchema) { - - private var _requiredMetadataSchema: StructType = StructType(Seq.empty) - - override def pruneColumns(requiredSchema: StructType): Unit = { - val resolver = sparkSession.sessionState.conf.resolver - val metaFields = requiredSchema.fields.filter { field => - !dataSchema.fields.exists(df => resolver(df.name, field.name)) && - !fileIndex.partitionSchema.fields.exists(pf => resolver(pf.name, field.name)) - } - _requiredMetadataSchema = StructType(metaFields) - super.pruneColumns(requiredSchema) - } - - override def build(): Scan = { - val fileIndexAdjusted = - if (loadOptions.showMetadata) - new InMemoryFileIndex( - sparkSession, - fileIndex.inputFiles.slice(0, 1).map(new org.apache.hadoop.fs.Path(_)), - options.asCaseSensitiveMap.asScala.toMap, - userDefinedSchema) - else fileIndex - - GeoPackageScan( - dataSchema, - sparkSession, - fileIndexAdjusted, - dataSchema, - readPartitionSchema(), - _requiredMetadataSchema, - options, - loadOptions) - } -} diff --git a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala b/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala deleted file mode 100644 index 078c9f23545..00000000000 --- a/spark/spark-4.0/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala +++ /dev/null @@ -1,119 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.hadoop.fs.FileStatus -import org.apache.sedona.sql.datasources.geopackage.connection.{FileSystemUtils, GeoPackageConnectionManager} -import org.apache.sedona.sql.datasources.geopackage.model.{GeoPackageOptions, MetadataSchema, TableType} -import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.connector.catalog.{MetadataColumn, SupportsMetadataColumns} -import org.apache.spark.sql.connector.read.ScanBuilder -import org.apache.spark.sql.connector.write.{LogicalWriteInfo, WriteBuilder} -import org.apache.spark.sql.execution.datasources.{FileFormat, PartitioningAwareFileIndex, SedonaFileIndexHelper} -import org.apache.spark.sql.execution.datasources.v2.FileTable -import org.apache.spark.sql.types.{DataType, DoubleType, IntegerType, LongType, StringType, StructField, StructType, TimestampType} -import org.apache.spark.sql.util.CaseInsensitiveStringMap -import org.apache.spark.util.SerializableConfiguration - -import scala.jdk.CollectionConverters._ - -case class GeoPackageTable( - name: String, - sparkSession: SparkSession, - options: CaseInsensitiveStringMap, - paths: Seq[String], - userSpecifiedSchema: Option[StructType], - fallbackFileFormat: Class[_ <: FileFormat], - loadOptions: GeoPackageOptions) - extends FileTable(sparkSession, options, paths, userSpecifiedSchema) - with SupportsMetadataColumns { - - // Override fileIndex to skip the FileStreamSink.hasMetadata check that causes - // spurious FileNotFoundException warnings when reading from cloud storage (e.g., S3). - // GeoPackage tables are always non-streaming batch sources, so the streaming - // metadata check is unnecessary. - override lazy val fileIndex: PartitioningAwareFileIndex = - SedonaFileIndexHelper.createFileIndex(sparkSession, options, paths, userSpecifiedSchema) - - override def inferSchema(files: Seq[FileStatus]): Option[StructType] = { - if (loadOptions.showMetadata) { - return MetadataSchema.schema - } - - val serializableConf = new SerializableConfiguration( - sparkSession.sessionState.newHadoopConfWithOptions(options.asScala.toMap)) - - val (tempFile, copied) = - FileSystemUtils.copyToLocal(serializableConf.value, files.head.getPath) - - if (copied) { - tempFile.deleteOnExit() - } - - val tableType = if (loadOptions.showMetadata) { - TableType.METADATA - } else { - GeoPackageConnectionManager.findFeatureMetadata(tempFile.getPath, loadOptions.tableName) - } - - Some( - StructType( - GeoPackageConnectionManager - .getSchema(tempFile.getPath, loadOptions.tableName) - .map(field => field.toStructField(tableType)))) - } - - override def formatName: String = { - "GeoPackage" - } - - override def metadataColumns(): Array[MetadataColumn] = GeoPackageTable.fileMetadataColumns - - override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = { - new GeoPackageScanBuilder( - sparkSession, - fileIndex, - schema, - options, - loadOptions, - userSpecifiedSchema) - } - - override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = { - null - } -} - -object GeoPackageTable { - - private val FILE_METADATA_STRUCT_TYPE: StructType = StructType( - Seq( - StructField("file_path", StringType, nullable = false), - StructField("file_name", StringType, nullable = false), - StructField("file_size", LongType, nullable = false), - StructField("file_block_start", LongType, nullable = false), - StructField("file_block_length", LongType, nullable = false), - StructField("file_modification_time", TimestampType, nullable = false))) - - private[geopackage] val fileMetadataColumns: Array[MetadataColumn] = Array(new MetadataColumn { - override def name: String = "_metadata" - override def dataType: DataType = FILE_METADATA_STRUCT_TYPE - override def isNullable: Boolean = false - }) -} diff --git a/spark/spark-4.0/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala b/spark/spark-4.0/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala deleted file mode 100644 index acd2a7093f4..00000000000 --- a/spark/spark-4.0/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala +++ /dev/null @@ -1,476 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql - -import io.minio.{MakeBucketArgs, MinioClient, PutObjectArgs} -import org.apache.spark.sql.{DataFrame, Row, SparkSession} -import org.apache.spark.sql.functions.expr -import org.apache.spark.sql.sedona_sql.UDT.GeometryUDT -import org.apache.spark.sql.types.{BinaryType, BooleanType, DateType, DoubleType, IntegerType, StringType, StructField, StructType, TimestampType} -import org.scalatest.matchers.should.Matchers -import org.scalatest.prop.TableDrivenPropertyChecks._ -import org.testcontainers.containers.MinIOContainer -import org.testcontainers.utility.DockerImageName - -import java.io.FileInputStream -import java.sql.{Date, Timestamp} -import java.util.TimeZone - -class GeoPackageReaderTest extends TestBaseScala with Matchers { - TimeZone.setDefault(TimeZone.getTimeZone("UTC")) - import sparkSession.implicits._ - - val path: String = resourceFolder + "geopackage/example.gpkg" - val polygonsPath: String = resourceFolder + "geopackage/features.gpkg" - val rasterPath: String = resourceFolder + "geopackage/raster.gpkg" - val wktReader = new org.locationtech.jts.io.WKTReader() - val wktWriter = new org.locationtech.jts.io.WKTWriter() - - val expectedFeatureSchema = StructType( - Seq( - StructField("id", IntegerType, true), - StructField("geometry", GeometryUDT(), true), - StructField("text", StringType, true), - StructField("real", DoubleType, true), - StructField("boolean", BooleanType, true), - StructField("blob", BinaryType, true), - StructField("integer", IntegerType, true), - StructField("text_limited", StringType, true), - StructField("blob_limited", BinaryType, true), - StructField("date", DateType, true), - StructField("datetime", TimestampType, true))) - - describe("Reading GeoPackage metadata") { - it("should read GeoPackage metadata") { - val df = sparkSession.read - .format("geopackage") - .option("showMetadata", "true") - .load(path) - - df.where("data_type = 'tiles'").show(false) - - df.count shouldEqual 34 - } - } - - describe("Reading Vector data") { - it("should read GeoPackage - point1") { - val df = readFeatureData("point1") - df.schema shouldEqual expectedFeatureSchema - - df.count() shouldEqual 4 - - val firstElement = df.collectAsList().get(0).toSeq - - val expectedValues = Seq( - 1, - wktReader.read(POINT_1), - "BIT Systems", - 4519.866024037493, - true, - Array(48, 99, 57, 54, 49, 56, 55, 54, 45, 98, 102, 100, 52, 45, 52, 102, 52, 48, 45, 97, - 49, 102, 101, 45, 55, 49, 55, 101, 57, 100, 50, 98, 48, 55, 98, 101), - 3, - "bcd5a36f-16dc-4385-87be-b40353848597", - Array(49, 50, 53, 50, 97, 99, 98, 52, 45, 57, 54, 54, 52, 45, 52, 101, 51, 50, 45, 57, 54, - 100, 101, 45, 56, 48, 54, 101, 101, 48, 101, 101, 49, 102, 57, 48), - Date.valueOf("2023-09-19"), - Timestamp.valueOf("2023-09-19 11:24:15.695")) - - firstElement should contain theSameElementsAs expectedValues - } - - it("should read GeoPackage - line1") { - val df = readFeatureData("line1") - .withColumn("datetime", expr("from_utc_timestamp(datetime, 'UTC')")) - - df.schema shouldEqual expectedFeatureSchema - - df.count() shouldEqual 3 - - val firstElement = df.collectAsList().get(0).toSeq - - firstElement should contain theSameElementsAs Seq( - 1, - wktReader.read(LINESTRING_1), - "East Lockheed Drive", - 1990.5159635296877, - false, - Array(54, 97, 98, 100, 98, 51, 97, 56, 45, 54, 53, 101, 48, 45, 52, 55, 48, 54, 45, 56, - 50, 52, 48, 45, 51, 57, 48, 55, 99, 50, 102, 102, 57, 48, 99, 55), - 1, - "13dd91dc-3b7d-4d8d-a0ca-b3afb8e31c3d", - Array(57, 54, 98, 102, 56, 99, 101, 56, 45, 102, 48, 54, 49, 45, 52, 55, 99, 48, 45, 97, - 98, 48, 101, 45, 97, 99, 50, 52, 100, 98, 50, 97, 102, 50, 50, 54), - Date.valueOf("2023-09-19"), - Timestamp.valueOf("2023-09-19 11:24:15.716")) - } - - it("should read GeoPackage - polygon1") { - val df = readFeatureData("polygon1") - df.count shouldEqual 3 - df.schema shouldEqual expectedFeatureSchema - - df.select("geometry").collectAsList().get(0).toSeq should contain theSameElementsAs Seq( - wktReader.read(POLYGON_1)) - } - - it("should read GeoPackage - geometry1") { - val df = readFeatureData("geometry1") - df.count shouldEqual 10 - df.schema shouldEqual expectedFeatureSchema - - df.selectExpr("ST_ASTEXT(geometry)") - .as[String] - .collect() should contain theSameElementsAs Seq( - POINT_1, - POINT_2, - POINT_3, - POINT_4, - LINESTRING_1, - LINESTRING_2, - LINESTRING_3, - POLYGON_1, - POLYGON_2, - POLYGON_3) - } - - it("should read polygon with envelope data") { - val tables = Table( - ("tableName", "expectedCount"), - ("GB_Hex_5km_GS_CompressibleGround_v8", 4233), - ("GB_Hex_5km_GS_Landslides_v8", 4228), - ("GB_Hex_5km_GS_RunningSand_v8", 4233), - ("GB_Hex_5km_GS_ShrinkSwell_v8", 4233), - ("GB_Hex_5km_GS_SolubleRocks_v8", 4295)) - - forAll(tables) { (tableName: String, expectedCount: Int) => - val df = sparkSession.read - .format("geopackage") - .option("tableName", tableName) - .load(polygonsPath) - - df.count() shouldEqual expectedCount - } - } - - it("should handle datetime fields without timezone information") { - // This test verifies the fix for DateTimeParseException when reading - // GeoPackage files with datetime fields that don't include timezone info - val testFilePath = resourceFolder + "geopackage/test_datetime_issue.gpkg" - - // Test reading the test_features table with problematic datetime formats - val df = sparkSession.read - .format("geopackage") - .option("tableName", "test_features") - .load(testFilePath) - - // The test should not throw DateTimeParseException when reading datetime fields - noException should be thrownBy { - df.select("created_at", "updated_at").collect() - } - - // Verify that datetime fields are properly parsed as TimestampType - df.schema.fields.find(_.name == "created_at").get.dataType shouldEqual TimestampType - df.schema.fields.find(_.name == "updated_at").get.dataType shouldEqual TimestampType - - // Verify that we can read the datetime values - val datetimeValues = df.select("created_at", "updated_at").collect() - datetimeValues should not be empty - - // Verify that datetime values are valid timestamps - datetimeValues.foreach { row => - val createdTimestamp = row.getAs[Timestamp]("created_at") - val updatedTimestamp = row.getAs[Timestamp]("updated_at") - createdTimestamp should not be null - updatedTimestamp should not be null - createdTimestamp.getTime should be > 0L - updatedTimestamp.getTime should be > 0L - } - - // Test showMetadata option with the same file - noException should be thrownBy { - val metadataDf = sparkSession.read - .format("geopackage") - .option("showMetadata", "true") - .load(testFilePath) - metadataDf.select("last_change").collect() - } - } - - it("should read NULL date and datetime values as null") { - // A NULL in a DATE or DATETIME column used to surface as a NullPointerException - // from DataTypesTransformations and abort the whole scan. - val df = sparkSession.read - .format("geopackage") - .option("tableName", "test_features") - .load(resourceFolder + "geopackage/test_null_datetime.gpkg") - - df.schema.fields.find(_.name == "event_date").get.dataType shouldEqual DateType - df.schema.fields.find(_.name == "event_time").get.dataType shouldEqual TimestampType - - val rows = df - .select("fid", "event_date", "event_time") - .collect() - .map { row => - val date = Option(row.getAs[Date]("event_date")).map(_.toLocalDate.toString).orNull - val time = Option(row.getAs[Timestamp]("event_time")).map(_.toInstant.toString).orNull - (row.getInt(0), date, time) - } - .sortBy(_._1) - - rows shouldEqual Array( - (1, "2024-01-15", "2024-01-15T10:30:00Z"), - (2, null, null), - (3, "2024-03-01", null), - (4, null, "2024-03-01T00:00:00Z")) - - df.filter("event_date IS NULL").count() shouldEqual 2 - df.filter("event_time IS NULL").count() shouldEqual 2 - } - } - - describe("GeoPackage Raster Data Test") { - it("should read") { - val fractions = - Table( - ("tableName", "channelNumber", "expectedSum"), - ("point1_tiles", 4, 466591.0), - ("line1_tiles", 4, 5775976.0), - ("polygon1_tiles", 4, 1.1269871e7), - ("geometry1_tiles", 4, 2.6328442e7), - ("point2_tiles", 4, 137456.0), - ("line2_tiles", 4, 6701101.0), - ("polygon2_tiles", 4, 5.1170714e7), - ("geometry2_tiles", 4, 1.6699823e7), - ("bit_systems", 1, 6.5561879e7), - ("nga", 1, 6.8078856e7), - ("bit_systems_wgs84", 1, 7.7276934e7), - ("nga_pc", 1, 2.90590616e8), - ("bit_systems_world", 1, 7.7276934e7), - ("nga_pc_world", 1, 2.90590616e8)) - - forAll(fractions) { (tableName: String, channelNumber: Int, expectedSum: Double) => - { - val df = readFeatureData(tableName) - val calculatedSum = df - .selectExpr(s"RS_SummaryStats(tile_data, 'sum', ${channelNumber}) as stats") - .selectExpr("sum(stats)") - .as[Double] - - calculatedSum.collect().head shouldEqual expectedSum - } - } - } - - it("should be able to read complex raster data") { - val df = sparkSession.read - .format("geopackage") - .option("tableName", "AuroraAirportNoise") - .load(rasterPath) - - df.show(5) - - val calculatedSum = df - .selectExpr(s"RS_SummaryStats(tile_data, 'sum', ${1}) as stats") - .selectExpr("sum(stats)") - .as[Double] - - calculatedSum.first() shouldEqual 2.027126e7 - - val df2 = sparkSession.read - .format("geopackage") - .option("tableName", "LiquorLicenseDensity") - .load(rasterPath) - - val calculatedSum2 = df2 - .selectExpr(s"RS_SummaryStats(tile_data, 'sum', ${1}) as stats") - .selectExpr("sum(stats)") - .as[Double] - - calculatedSum2.first() shouldEqual 2.882028e7 - } - - } - - describe("Reading from S3") { - it("should be able to read files from S3") { - val container = new MinIOContainer( - DockerImageName - .parse("quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z") - .asCompatibleSubstituteFor("minio/minio")) - - container.start() - - val minioClient = createMinioClient(container) - val makeBucketRequest = MakeBucketArgs - .builder() - .bucket("sedona") - .build() - - minioClient.makeBucket(makeBucketRequest) - - adjustSparkSession(sparkSessionMinio, container) - - val inputPath: String = prepareFile("example.geopackage", path, minioClient) - - sparkSessionMinio.read - .format("geopackage") - .option("showMetadata", "true") - .load(inputPath) - .count shouldEqual 34 - - val df = sparkSession.read - .format("geopackage") - .option("tableName", "point1") - .load(inputPath) - - df.count shouldEqual 4 - - val inputPathLarger: String = prepareFiles((1 to 300).map(_ => path).toArray, minioClient) - - val dfLarger = sparkSessionMinio.read - .format("geopackage") - .option("tableName", "point1") - .load(inputPathLarger) - - dfLarger.count shouldEqual 300 * 4 - - container.stop() - } - } - - describe("_metadata hidden column support") { - it("should expose _metadata struct with all expected fields") { - val df = readFeatureData("point1") - val metaDf = df.select("_metadata") - val metaSchema = metaDf.schema.fields.head.dataType.asInstanceOf[StructType] - val fieldNames = metaSchema.fieldNames.toSet - fieldNames should contain("file_path") - fieldNames should contain("file_name") - fieldNames should contain("file_size") - fieldNames should contain("file_block_start") - fieldNames should contain("file_block_length") - fieldNames should contain("file_modification_time") - } - - it("should not include _metadata in select(*)") { - val df = readFeatureData("point1") - val starCols = df.select("*").columns.toSet - starCols should not contain "_metadata" - } - - it("should return correct file_path and file_name in _metadata") { - val df = readFeatureData("point1") - val row = df.select("_metadata.file_path", "_metadata.file_name").head() - val filePath = row.getString(0) - val fileName = row.getString(1) - filePath should endWith("example.gpkg") - fileName shouldEqual "example.gpkg" - } - - it("should return actual file_size matching the .gpkg file on disk") { - val df = readFeatureData("point1") - val metaFileSize = df.select("_metadata.file_size").head().getLong(0) - val actualFile = new java.io.File(path) - metaFileSize shouldEqual actualFile.length() - } - - it("should return file_block_start=0 and file_block_length=file_size") { - val df = readFeatureData("point1") - val row = df - .select( - "_metadata.file_block_start", - "_metadata.file_block_length", - "_metadata.file_size") - .head() - row.getLong(0) shouldEqual 0L - row.getLong(1) shouldEqual row.getLong(2) - } - - it("should return file_modification_time matching the .gpkg file on disk") { - val df = readFeatureData("point1") - val metaModTime = df.select("_metadata.file_modification_time").head().getTimestamp(0) - val actualFile = new java.io.File(path) - val expectedModTime = new java.sql.Timestamp(actualFile.lastModified()) - metaModTime shouldEqual expectedModTime - } - - it("should allow filtering on _metadata fields") { - val df = readFeatureData("point1") - val filtered = df.filter(df("_metadata.file_name") === "example.gpkg") - filtered.count() shouldEqual df.count() - val empty = df.filter(df("_metadata.file_name") === "nonexistent.gpkg") - empty.count() shouldEqual 0 - } - - it("should select _metadata along with data columns") { - val df = readFeatureData("point1") - val result = df.select("id", "_metadata.file_name").head() - result.getInt(0) shouldEqual 1 - result.getString(1) shouldEqual "example.gpkg" - } - } - - private def readFeatureData(tableName: String): DataFrame = { - sparkSession.read - .format("geopackage") - .option("tableName", tableName) - .load(path) - } - - private def prepareFiles(paths: Array[String], minioClient: MinioClient): String = { - val key = "geopackage" - - paths.foreach(path => { - val fis = new FileInputStream(path); - putFileIntoBucket( - "sedona", - s"${key}/${scala.util.Random.nextInt(1000000000)}.geopackage", - fis, - minioClient) - }) - - s"s3a://sedona/$key" - } - - private def prepareFile(name: String, path: String, minioClient: MinioClient): String = { - val fis = new FileInputStream(path); - putFileIntoBucket("sedona", name, fis, minioClient) - - s"s3a://sedona/$name" - } - - private val POINT_1 = "POINT (-104.801918 39.720014)" - private val POINT_2 = "POINT (-104.802987 39.717703)" - private val POINT_3 = "POINT (-104.807496 39.714085)" - private val POINT_4 = "POINT (-104.79948 39.714729)" - private val LINESTRING_1 = - "LINESTRING (-104.800614 39.720721, -104.802174 39.720726, -104.802584 39.72066, -104.803088 39.720477, -104.803474 39.720209)" - private val LINESTRING_2 = - "LINESTRING (-104.809612 39.718379, -104.806638 39.718372, -104.806236 39.718439, -104.805939 39.718536, -104.805654 39.718677, -104.803652 39.720095)" - private val LINESTRING_3 = - "LINESTRING (-104.806344 39.722425, -104.805854 39.722634, -104.805656 39.722647, -104.803749 39.722641, -104.803769 39.721849, -104.803806 39.721725, -104.804382 39.720865)" - private val POLYGON_1 = - "POLYGON ((-104.802246 39.720343, -104.802246 39.719753, -104.802183 39.719754, -104.802184 39.719719, -104.802138 39.719694, -104.802097 39.719691, -104.802096 39.719648, -104.801646 39.719648, -104.801644 39.719722, -104.80155 39.719723, -104.801549 39.720207, -104.801648 39.720207, -104.801648 39.720341, -104.802246 39.720343))" - private val POLYGON_2 = - "POLYGON ((-104.802259 39.719604, -104.80226 39.71955, -104.802281 39.719416, -104.802332 39.719372, -104.802081 39.71924, -104.802044 39.71929, -104.802027 39.719278, -104.802044 39.719229, -104.801785 39.719129, -104.801639 39.719413, -104.801649 39.719472, -104.801694 39.719524, -104.801753 39.71955, -104.80175 39.719606, -104.80194 39.719606, -104.801939 39.719555, -104.801977 39.719556, -104.801979 39.719606, -104.802259 39.719604), (-104.80213 39.71944, -104.802133 39.71949, -104.802148 39.71949, -104.80218 39.719473, -104.802187 39.719456, -104.802182 39.719439, -104.802088 39.719387, -104.802047 39.719427, -104.801858 39.719342, -104.801883 39.719294, -104.801832 39.719284, -104.801787 39.719298, -104.801763 39.719331, -104.801823 39.719352, -104.80179 39.71942, -104.801722 39.719404, -104.801715 39.719445, -104.801748 39.719484, -104.801809 39.719494, -104.801816 39.719439, -104.80213 39.71944))" - private val POLYGON_3 = - "POLYGON ((-104.802867 39.718122, -104.802369 39.717845, -104.802571 39.71763, -104.803066 39.717909, -104.802867 39.718122))" -} diff --git a/spark/spark-4.1/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister b/spark/spark-4.1/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister index ae1de3d8bd2..c9a3961320e 100644 --- a/spark/spark-4.1/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister +++ b/spark/spark-4.1/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister @@ -1,3 +1,2 @@ org.apache.spark.sql.execution.datasources.v2.geoparquet.metadata.GeoParquetMetadataDataSource org.apache.sedona.sql.datasources.shapefile.ShapefileDataSource -org.apache.sedona.sql.datasources.geopackage.GeoPackageDataSource diff --git a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala b/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala deleted file mode 100644 index 11f2db38e84..00000000000 --- a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageDataSource.scala +++ /dev/null @@ -1,73 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.hadoop.fs.Path -import org.apache.sedona.sql.datasources.geopackage.model.GeoPackageOptions -import org.apache.spark.sql.connector.catalog.Table -import org.apache.spark.sql.execution.datasources.FileFormat -import org.apache.spark.sql.execution.datasources.v2.FileDataSourceV2 -import org.apache.spark.sql.sources.DataSourceRegister -import org.apache.spark.sql.util.CaseInsensitiveStringMap - -import java.util.Locale -import scala.jdk.CollectionConverters._ -import scala.util.Try - -class GeoPackageDataSource extends FileDataSourceV2 with DataSourceRegister { - - override def fallbackFileFormat: Class[_ <: FileFormat] = { - null - } - - override protected def getTable(options: CaseInsensitiveStringMap): Table = { - GeoPackageTable( - "", - sparkSession, - options, - getPaths(options), - None, - fallbackFileFormat, - getLoadOptions(options)) - } - - private def getLoadOptions(options: CaseInsensitiveStringMap): GeoPackageOptions = { - val path = options.get("path") - if (path.isEmpty) { - throw new IllegalArgumentException("GeoPackage path is not specified") - } - - val showMetadata = options.getBoolean("showMetadata", false) - val maybeTableName = options.get("tableName") - - if (!showMetadata && maybeTableName == null) { - throw new IllegalArgumentException("Table name is not specified") - } - - val tableName = if (showMetadata) { - "gpkg_contents" - } else { - maybeTableName - } - - GeoPackageOptions(tableName = tableName, showMetadata = showMetadata) - } - - override def shortName(): String = "geopackage" -} diff --git a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala b/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala deleted file mode 100644 index 4e59163922d..00000000000 --- a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReader.scala +++ /dev/null @@ -1,107 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.hadoop.fs.Path -import org.apache.sedona.sql.datasources.geopackage.connection.{FileSystemUtils, GeoPackageConnectionManager} -import org.apache.sedona.sql.datasources.geopackage.model.TableType.{FEATURES, METADATA, TILES, UNKNOWN} -import org.apache.sedona.sql.datasources.geopackage.model.{GeoPackageReadOptions, PartitionOptions, TileRowMetadata} -import org.apache.sedona.sql.datasources.geopackage.transform.ValuesMapper -import org.apache.spark.broadcast.Broadcast -import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.connector.read.PartitionReader -import org.apache.spark.util.SerializableConfiguration - -import java.io.File -import java.sql.ResultSet - -case class GeoPackagePartitionReader( - var rs: ResultSet, - options: GeoPackageReadOptions, - broadcastedConf: Broadcast[SerializableConfiguration], - var currentTempFile: File, - copying: Boolean = false) - extends PartitionReader[InternalRow] { - - private var values: Seq[Any] = Seq.empty - private var currentFile = options.currentFile - private val partitionedFiles = options.partitionedFiles - - override def next(): Boolean = { - if (rs.next()) { - values = ValuesMapper.mapValues(adjustPartitionOptions, rs) - return true - } - - partitionedFiles.remove(currentFile) - - if (partitionedFiles.isEmpty) { - return false - } - - rs.close() - - currentFile = partitionedFiles.head - val (tempFile, _) = FileSystemUtils.copyToLocal( - options = broadcastedConf.value.value, - file = new Path(currentFile.filePath.toString())) - - if (copying) { - currentTempFile.deleteOnExit() - } - - currentTempFile = tempFile - - rs = GeoPackageConnectionManager.getTableCursor(currentTempFile.getPath, options.tableName) - - if (!rs.next()) { - return false - } - - values = ValuesMapper.mapValues(adjustPartitionOptions, rs) - - true - } - - private def adjustPartitionOptions: PartitionOptions = { - options.partitionOptions.tableType match { - case FEATURES | METADATA => options.partitionOptions - case TILES => - val tileRowMetadata = TileRowMetadata( - zoomLevel = rs.getInt("zoom_level"), - tileColumn = rs.getInt("tile_column"), - tileRow = rs.getInt("tile_row")) - - options.partitionOptions.withTileRowMetadata(tileRowMetadata) - case UNKNOWN => options.partitionOptions - } - - } - - override def get(): InternalRow = { - InternalRow.fromSeq(values) - } - - override def close(): Unit = { - rs.close() - if (copying) { - options.tempFile.delete() - } - } -} diff --git a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala b/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala deleted file mode 100644 index 0f2e9a87b81..00000000000 --- a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackagePartitionReaderFactory.scala +++ /dev/null @@ -1,139 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.hadoop.fs.Path -import org.apache.sedona.sql.datasources.geopackage.connection.{FileSystemUtils, GeoPackageConnectionManager} -import org.apache.sedona.sql.datasources.geopackage.model.TableType.TILES -import org.apache.sedona.sql.datasources.geopackage.model.{GeoPackageOptions, GeoPackageReadOptions, PartitionOptions, TableType} -import org.apache.spark.broadcast.Broadcast -import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.catalyst.expressions.{BoundReference, JoinedRow} -import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection -import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader, PartitionReaderFactory} -import org.apache.spark.sql.execution.datasources.FilePartition -import org.apache.spark.sql.types.StructType -import org.apache.spark.unsafe.types.UTF8String -import org.apache.spark.util.SerializableConfiguration - -case class GeoPackagePartitionReaderFactory( - sparkSession: SparkSession, - broadcastedConf: Broadcast[SerializableConfiguration], - loadOptions: GeoPackageOptions, - dataSchema: StructType, - metadataSchema: StructType) - extends PartitionReaderFactory { - - override def createReader(partition: InputPartition): PartitionReader[InternalRow] = { - val partitionFiles = partition match { - case filePartition: FilePartition => filePartition.files - case _ => - throw new IllegalArgumentException( - s"Unexpected partition type: ${partition.getClass.getCanonicalName}") - } - - val (tempFile, copied) = FileSystemUtils.copyToLocal( - options = broadcastedConf.value.value, - file = new Path(partitionFiles.head.filePath.toString())) - - val tableType = if (loadOptions.showMetadata) { - TableType.METADATA - } else { - GeoPackageConnectionManager.findFeatureMetadata(tempFile.getPath, loadOptions.tableName) - } - - val rs = - GeoPackageConnectionManager.getTableCursor(tempFile.getAbsolutePath, loadOptions.tableName) - - val schema = GeoPackageConnectionManager.getSchema(tempFile.getPath, loadOptions.tableName) - - if (StructType(schema.map(_.toStructField(tableType))) != dataSchema) { - throw new IllegalArgumentException( - s"Schema mismatch: expected $dataSchema, got ${StructType(schema.map(_.toStructField(tableType)))}") - } - - val tileMetadata = tableType match { - case TILES => - Some( - GeoPackageConnectionManager.findTilesMetadata(tempFile.getPath, loadOptions.tableName)) - case _ => None - } - - val baseReader = GeoPackagePartitionReader( - rs = rs, - options = GeoPackageReadOptions( - tableName = loadOptions.tableName, - tempFile = tempFile, - partitionOptions = - PartitionOptions(tableType = tableType, columns = schema, tile = tileMetadata), - partitionedFiles = scala.collection.mutable.HashSet(partitionFiles: _*), - currentFile = partitionFiles.head), - broadcastedConf = broadcastedConf, - currentTempFile = tempFile, - copying = copied) - - if (metadataSchema.nonEmpty) { - val gpkgFile = partitionFiles.head - val filePath = gpkgFile.filePath.toString - val fileName = new Path(filePath).getName - - val allMetadataValues: Map[String, Any] = Map( - "file_path" -> UTF8String.fromString(filePath), - "file_name" -> UTF8String.fromString(fileName), - "file_size" -> gpkgFile.fileSize, - "file_block_start" -> gpkgFile.start, - "file_block_length" -> gpkgFile.length, - "file_modification_time" -> (gpkgFile.modificationTime * 1000L)) - - val innerStructType = metadataSchema.fields.head.dataType.asInstanceOf[StructType] - val prunedValues = innerStructType.fields.map(f => allMetadataValues(f.name)) - val metadataStruct = InternalRow.fromSeq(prunedValues.toSeq) - val metadataRow = InternalRow.fromSeq(Seq(metadataStruct)) - - new PartitionReaderWithMetadata(baseReader, dataSchema, metadataSchema, metadataRow) - } else { - baseReader - } - } -} - -private[geopackage] class PartitionReaderWithMetadata( - reader: PartitionReader[InternalRow], - baseSchema: StructType, - metadataSchema: StructType, - metadataValues: InternalRow) - extends PartitionReader[InternalRow] { - - private val joinedRow = new JoinedRow() - private val unsafeProjection = - GenerateUnsafeProjection.generate(baseSchema.fields.zipWithIndex.map { case (f, i) => - BoundReference(i, f.dataType, f.nullable) - } ++ metadataSchema.fields.zipWithIndex.map { case (f, i) => - BoundReference(baseSchema.length + i, f.dataType, f.nullable) - }) - - override def next(): Boolean = reader.next() - - override def get(): InternalRow = { - unsafeProjection(joinedRow(reader.get(), metadataValues)) - } - - override def close(): Unit = reader.close() -} diff --git a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala b/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala deleted file mode 100644 index edca3d35ff1..00000000000 --- a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScan.scala +++ /dev/null @@ -1,68 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.sedona.sql.datasources.geopackage.model.GeoPackageOptions -import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.catalyst.expressions.Expression -import org.apache.spark.sql.connector.read.PartitionReaderFactory -import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex -import org.apache.spark.sql.execution.datasources.v2.FileScan -import org.apache.spark.sql.types.StructType -import org.apache.spark.sql.util.CaseInsensitiveStringMap -import org.apache.spark.util.SerializableConfiguration - -import scala.jdk.CollectionConverters._ - -case class GeoPackageScan( - dataSchema: StructType, - sparkSession: SparkSession, - fileIndex: PartitioningAwareFileIndex, - readDataSchema: StructType, - readPartitionSchema: StructType, - metadataSchema: StructType, - options: CaseInsensitiveStringMap, - loadOptions: GeoPackageOptions) - extends FileScan { - - override def readSchema(): StructType = - StructType(readDataSchema.fields ++ readPartitionSchema.fields ++ metadataSchema.fields) - - override def partitionFilters: Seq[Expression] = { - Seq.empty - } - - override def dataFilters: Seq[Expression] = { - Seq.empty - } - - override def createReaderFactory(): PartitionReaderFactory = { - val caseSensitiveMap = options.asScala.toMap - val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(caseSensitiveMap) - val broadcastedConf = - sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf)) - - GeoPackagePartitionReaderFactory( - sparkSession, - broadcastedConf, - loadOptions, - dataSchema, - metadataSchema) - } -} diff --git a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala b/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala deleted file mode 100644 index 7fdb716f29f..00000000000 --- a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageScanBuilder.scala +++ /dev/null @@ -1,71 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.sedona.sql.datasources.geopackage.model.GeoPackageOptions -import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.connector.read.Scan -import org.apache.spark.sql.execution.datasources.{InMemoryFileIndex, PartitioningAwareFileIndex} -import org.apache.spark.sql.execution.datasources.v2.FileScanBuilder -import org.apache.spark.sql.types.StructType -import org.apache.spark.sql.util.CaseInsensitiveStringMap -import scala.jdk.CollectionConverters._ - -class GeoPackageScanBuilder( - sparkSession: SparkSession, - fileIndex: PartitioningAwareFileIndex, - dataSchema: StructType, - options: CaseInsensitiveStringMap, - loadOptions: GeoPackageOptions, - userDefinedSchema: Option[StructType] = None) - extends FileScanBuilder(sparkSession, fileIndex, dataSchema) { - - private var _requiredMetadataSchema: StructType = StructType(Seq.empty) - - override def pruneColumns(requiredSchema: StructType): Unit = { - val resolver = sparkSession.sessionState.conf.resolver - val metaFields = requiredSchema.fields.filter { field => - !dataSchema.fields.exists(df => resolver(df.name, field.name)) && - !fileIndex.partitionSchema.fields.exists(pf => resolver(pf.name, field.name)) - } - _requiredMetadataSchema = StructType(metaFields) - super.pruneColumns(requiredSchema) - } - - override def build(): Scan = { - val fileIndexAdjusted = - if (loadOptions.showMetadata) - new InMemoryFileIndex( - sparkSession, - fileIndex.inputFiles.slice(0, 1).map(new org.apache.hadoop.fs.Path(_)), - options.asCaseSensitiveMap.asScala.toMap, - userDefinedSchema) - else fileIndex - - GeoPackageScan( - dataSchema, - sparkSession, - fileIndexAdjusted, - dataSchema, - readPartitionSchema(), - _requiredMetadataSchema, - options, - loadOptions) - } -} diff --git a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala b/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala deleted file mode 100644 index 078c9f23545..00000000000 --- a/spark/spark-4.1/src/main/scala/org/apache/sedona/sql/datasources/geopackage/GeoPackageTable.scala +++ /dev/null @@ -1,119 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql.datasources.geopackage - -import org.apache.hadoop.fs.FileStatus -import org.apache.sedona.sql.datasources.geopackage.connection.{FileSystemUtils, GeoPackageConnectionManager} -import org.apache.sedona.sql.datasources.geopackage.model.{GeoPackageOptions, MetadataSchema, TableType} -import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.connector.catalog.{MetadataColumn, SupportsMetadataColumns} -import org.apache.spark.sql.connector.read.ScanBuilder -import org.apache.spark.sql.connector.write.{LogicalWriteInfo, WriteBuilder} -import org.apache.spark.sql.execution.datasources.{FileFormat, PartitioningAwareFileIndex, SedonaFileIndexHelper} -import org.apache.spark.sql.execution.datasources.v2.FileTable -import org.apache.spark.sql.types.{DataType, DoubleType, IntegerType, LongType, StringType, StructField, StructType, TimestampType} -import org.apache.spark.sql.util.CaseInsensitiveStringMap -import org.apache.spark.util.SerializableConfiguration - -import scala.jdk.CollectionConverters._ - -case class GeoPackageTable( - name: String, - sparkSession: SparkSession, - options: CaseInsensitiveStringMap, - paths: Seq[String], - userSpecifiedSchema: Option[StructType], - fallbackFileFormat: Class[_ <: FileFormat], - loadOptions: GeoPackageOptions) - extends FileTable(sparkSession, options, paths, userSpecifiedSchema) - with SupportsMetadataColumns { - - // Override fileIndex to skip the FileStreamSink.hasMetadata check that causes - // spurious FileNotFoundException warnings when reading from cloud storage (e.g., S3). - // GeoPackage tables are always non-streaming batch sources, so the streaming - // metadata check is unnecessary. - override lazy val fileIndex: PartitioningAwareFileIndex = - SedonaFileIndexHelper.createFileIndex(sparkSession, options, paths, userSpecifiedSchema) - - override def inferSchema(files: Seq[FileStatus]): Option[StructType] = { - if (loadOptions.showMetadata) { - return MetadataSchema.schema - } - - val serializableConf = new SerializableConfiguration( - sparkSession.sessionState.newHadoopConfWithOptions(options.asScala.toMap)) - - val (tempFile, copied) = - FileSystemUtils.copyToLocal(serializableConf.value, files.head.getPath) - - if (copied) { - tempFile.deleteOnExit() - } - - val tableType = if (loadOptions.showMetadata) { - TableType.METADATA - } else { - GeoPackageConnectionManager.findFeatureMetadata(tempFile.getPath, loadOptions.tableName) - } - - Some( - StructType( - GeoPackageConnectionManager - .getSchema(tempFile.getPath, loadOptions.tableName) - .map(field => field.toStructField(tableType)))) - } - - override def formatName: String = { - "GeoPackage" - } - - override def metadataColumns(): Array[MetadataColumn] = GeoPackageTable.fileMetadataColumns - - override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = { - new GeoPackageScanBuilder( - sparkSession, - fileIndex, - schema, - options, - loadOptions, - userSpecifiedSchema) - } - - override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = { - null - } -} - -object GeoPackageTable { - - private val FILE_METADATA_STRUCT_TYPE: StructType = StructType( - Seq( - StructField("file_path", StringType, nullable = false), - StructField("file_name", StringType, nullable = false), - StructField("file_size", LongType, nullable = false), - StructField("file_block_start", LongType, nullable = false), - StructField("file_block_length", LongType, nullable = false), - StructField("file_modification_time", TimestampType, nullable = false))) - - private[geopackage] val fileMetadataColumns: Array[MetadataColumn] = Array(new MetadataColumn { - override def name: String = "_metadata" - override def dataType: DataType = FILE_METADATA_STRUCT_TYPE - override def isNullable: Boolean = false - }) -} diff --git a/spark/spark-4.1/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala b/spark/spark-4.1/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala deleted file mode 100644 index acd2a7093f4..00000000000 --- a/spark/spark-4.1/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala +++ /dev/null @@ -1,476 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sedona.sql - -import io.minio.{MakeBucketArgs, MinioClient, PutObjectArgs} -import org.apache.spark.sql.{DataFrame, Row, SparkSession} -import org.apache.spark.sql.functions.expr -import org.apache.spark.sql.sedona_sql.UDT.GeometryUDT -import org.apache.spark.sql.types.{BinaryType, BooleanType, DateType, DoubleType, IntegerType, StringType, StructField, StructType, TimestampType} -import org.scalatest.matchers.should.Matchers -import org.scalatest.prop.TableDrivenPropertyChecks._ -import org.testcontainers.containers.MinIOContainer -import org.testcontainers.utility.DockerImageName - -import java.io.FileInputStream -import java.sql.{Date, Timestamp} -import java.util.TimeZone - -class GeoPackageReaderTest extends TestBaseScala with Matchers { - TimeZone.setDefault(TimeZone.getTimeZone("UTC")) - import sparkSession.implicits._ - - val path: String = resourceFolder + "geopackage/example.gpkg" - val polygonsPath: String = resourceFolder + "geopackage/features.gpkg" - val rasterPath: String = resourceFolder + "geopackage/raster.gpkg" - val wktReader = new org.locationtech.jts.io.WKTReader() - val wktWriter = new org.locationtech.jts.io.WKTWriter() - - val expectedFeatureSchema = StructType( - Seq( - StructField("id", IntegerType, true), - StructField("geometry", GeometryUDT(), true), - StructField("text", StringType, true), - StructField("real", DoubleType, true), - StructField("boolean", BooleanType, true), - StructField("blob", BinaryType, true), - StructField("integer", IntegerType, true), - StructField("text_limited", StringType, true), - StructField("blob_limited", BinaryType, true), - StructField("date", DateType, true), - StructField("datetime", TimestampType, true))) - - describe("Reading GeoPackage metadata") { - it("should read GeoPackage metadata") { - val df = sparkSession.read - .format("geopackage") - .option("showMetadata", "true") - .load(path) - - df.where("data_type = 'tiles'").show(false) - - df.count shouldEqual 34 - } - } - - describe("Reading Vector data") { - it("should read GeoPackage - point1") { - val df = readFeatureData("point1") - df.schema shouldEqual expectedFeatureSchema - - df.count() shouldEqual 4 - - val firstElement = df.collectAsList().get(0).toSeq - - val expectedValues = Seq( - 1, - wktReader.read(POINT_1), - "BIT Systems", - 4519.866024037493, - true, - Array(48, 99, 57, 54, 49, 56, 55, 54, 45, 98, 102, 100, 52, 45, 52, 102, 52, 48, 45, 97, - 49, 102, 101, 45, 55, 49, 55, 101, 57, 100, 50, 98, 48, 55, 98, 101), - 3, - "bcd5a36f-16dc-4385-87be-b40353848597", - Array(49, 50, 53, 50, 97, 99, 98, 52, 45, 57, 54, 54, 52, 45, 52, 101, 51, 50, 45, 57, 54, - 100, 101, 45, 56, 48, 54, 101, 101, 48, 101, 101, 49, 102, 57, 48), - Date.valueOf("2023-09-19"), - Timestamp.valueOf("2023-09-19 11:24:15.695")) - - firstElement should contain theSameElementsAs expectedValues - } - - it("should read GeoPackage - line1") { - val df = readFeatureData("line1") - .withColumn("datetime", expr("from_utc_timestamp(datetime, 'UTC')")) - - df.schema shouldEqual expectedFeatureSchema - - df.count() shouldEqual 3 - - val firstElement = df.collectAsList().get(0).toSeq - - firstElement should contain theSameElementsAs Seq( - 1, - wktReader.read(LINESTRING_1), - "East Lockheed Drive", - 1990.5159635296877, - false, - Array(54, 97, 98, 100, 98, 51, 97, 56, 45, 54, 53, 101, 48, 45, 52, 55, 48, 54, 45, 56, - 50, 52, 48, 45, 51, 57, 48, 55, 99, 50, 102, 102, 57, 48, 99, 55), - 1, - "13dd91dc-3b7d-4d8d-a0ca-b3afb8e31c3d", - Array(57, 54, 98, 102, 56, 99, 101, 56, 45, 102, 48, 54, 49, 45, 52, 55, 99, 48, 45, 97, - 98, 48, 101, 45, 97, 99, 50, 52, 100, 98, 50, 97, 102, 50, 50, 54), - Date.valueOf("2023-09-19"), - Timestamp.valueOf("2023-09-19 11:24:15.716")) - } - - it("should read GeoPackage - polygon1") { - val df = readFeatureData("polygon1") - df.count shouldEqual 3 - df.schema shouldEqual expectedFeatureSchema - - df.select("geometry").collectAsList().get(0).toSeq should contain theSameElementsAs Seq( - wktReader.read(POLYGON_1)) - } - - it("should read GeoPackage - geometry1") { - val df = readFeatureData("geometry1") - df.count shouldEqual 10 - df.schema shouldEqual expectedFeatureSchema - - df.selectExpr("ST_ASTEXT(geometry)") - .as[String] - .collect() should contain theSameElementsAs Seq( - POINT_1, - POINT_2, - POINT_3, - POINT_4, - LINESTRING_1, - LINESTRING_2, - LINESTRING_3, - POLYGON_1, - POLYGON_2, - POLYGON_3) - } - - it("should read polygon with envelope data") { - val tables = Table( - ("tableName", "expectedCount"), - ("GB_Hex_5km_GS_CompressibleGround_v8", 4233), - ("GB_Hex_5km_GS_Landslides_v8", 4228), - ("GB_Hex_5km_GS_RunningSand_v8", 4233), - ("GB_Hex_5km_GS_ShrinkSwell_v8", 4233), - ("GB_Hex_5km_GS_SolubleRocks_v8", 4295)) - - forAll(tables) { (tableName: String, expectedCount: Int) => - val df = sparkSession.read - .format("geopackage") - .option("tableName", tableName) - .load(polygonsPath) - - df.count() shouldEqual expectedCount - } - } - - it("should handle datetime fields without timezone information") { - // This test verifies the fix for DateTimeParseException when reading - // GeoPackage files with datetime fields that don't include timezone info - val testFilePath = resourceFolder + "geopackage/test_datetime_issue.gpkg" - - // Test reading the test_features table with problematic datetime formats - val df = sparkSession.read - .format("geopackage") - .option("tableName", "test_features") - .load(testFilePath) - - // The test should not throw DateTimeParseException when reading datetime fields - noException should be thrownBy { - df.select("created_at", "updated_at").collect() - } - - // Verify that datetime fields are properly parsed as TimestampType - df.schema.fields.find(_.name == "created_at").get.dataType shouldEqual TimestampType - df.schema.fields.find(_.name == "updated_at").get.dataType shouldEqual TimestampType - - // Verify that we can read the datetime values - val datetimeValues = df.select("created_at", "updated_at").collect() - datetimeValues should not be empty - - // Verify that datetime values are valid timestamps - datetimeValues.foreach { row => - val createdTimestamp = row.getAs[Timestamp]("created_at") - val updatedTimestamp = row.getAs[Timestamp]("updated_at") - createdTimestamp should not be null - updatedTimestamp should not be null - createdTimestamp.getTime should be > 0L - updatedTimestamp.getTime should be > 0L - } - - // Test showMetadata option with the same file - noException should be thrownBy { - val metadataDf = sparkSession.read - .format("geopackage") - .option("showMetadata", "true") - .load(testFilePath) - metadataDf.select("last_change").collect() - } - } - - it("should read NULL date and datetime values as null") { - // A NULL in a DATE or DATETIME column used to surface as a NullPointerException - // from DataTypesTransformations and abort the whole scan. - val df = sparkSession.read - .format("geopackage") - .option("tableName", "test_features") - .load(resourceFolder + "geopackage/test_null_datetime.gpkg") - - df.schema.fields.find(_.name == "event_date").get.dataType shouldEqual DateType - df.schema.fields.find(_.name == "event_time").get.dataType shouldEqual TimestampType - - val rows = df - .select("fid", "event_date", "event_time") - .collect() - .map { row => - val date = Option(row.getAs[Date]("event_date")).map(_.toLocalDate.toString).orNull - val time = Option(row.getAs[Timestamp]("event_time")).map(_.toInstant.toString).orNull - (row.getInt(0), date, time) - } - .sortBy(_._1) - - rows shouldEqual Array( - (1, "2024-01-15", "2024-01-15T10:30:00Z"), - (2, null, null), - (3, "2024-03-01", null), - (4, null, "2024-03-01T00:00:00Z")) - - df.filter("event_date IS NULL").count() shouldEqual 2 - df.filter("event_time IS NULL").count() shouldEqual 2 - } - } - - describe("GeoPackage Raster Data Test") { - it("should read") { - val fractions = - Table( - ("tableName", "channelNumber", "expectedSum"), - ("point1_tiles", 4, 466591.0), - ("line1_tiles", 4, 5775976.0), - ("polygon1_tiles", 4, 1.1269871e7), - ("geometry1_tiles", 4, 2.6328442e7), - ("point2_tiles", 4, 137456.0), - ("line2_tiles", 4, 6701101.0), - ("polygon2_tiles", 4, 5.1170714e7), - ("geometry2_tiles", 4, 1.6699823e7), - ("bit_systems", 1, 6.5561879e7), - ("nga", 1, 6.8078856e7), - ("bit_systems_wgs84", 1, 7.7276934e7), - ("nga_pc", 1, 2.90590616e8), - ("bit_systems_world", 1, 7.7276934e7), - ("nga_pc_world", 1, 2.90590616e8)) - - forAll(fractions) { (tableName: String, channelNumber: Int, expectedSum: Double) => - { - val df = readFeatureData(tableName) - val calculatedSum = df - .selectExpr(s"RS_SummaryStats(tile_data, 'sum', ${channelNumber}) as stats") - .selectExpr("sum(stats)") - .as[Double] - - calculatedSum.collect().head shouldEqual expectedSum - } - } - } - - it("should be able to read complex raster data") { - val df = sparkSession.read - .format("geopackage") - .option("tableName", "AuroraAirportNoise") - .load(rasterPath) - - df.show(5) - - val calculatedSum = df - .selectExpr(s"RS_SummaryStats(tile_data, 'sum', ${1}) as stats") - .selectExpr("sum(stats)") - .as[Double] - - calculatedSum.first() shouldEqual 2.027126e7 - - val df2 = sparkSession.read - .format("geopackage") - .option("tableName", "LiquorLicenseDensity") - .load(rasterPath) - - val calculatedSum2 = df2 - .selectExpr(s"RS_SummaryStats(tile_data, 'sum', ${1}) as stats") - .selectExpr("sum(stats)") - .as[Double] - - calculatedSum2.first() shouldEqual 2.882028e7 - } - - } - - describe("Reading from S3") { - it("should be able to read files from S3") { - val container = new MinIOContainer( - DockerImageName - .parse("quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z") - .asCompatibleSubstituteFor("minio/minio")) - - container.start() - - val minioClient = createMinioClient(container) - val makeBucketRequest = MakeBucketArgs - .builder() - .bucket("sedona") - .build() - - minioClient.makeBucket(makeBucketRequest) - - adjustSparkSession(sparkSessionMinio, container) - - val inputPath: String = prepareFile("example.geopackage", path, minioClient) - - sparkSessionMinio.read - .format("geopackage") - .option("showMetadata", "true") - .load(inputPath) - .count shouldEqual 34 - - val df = sparkSession.read - .format("geopackage") - .option("tableName", "point1") - .load(inputPath) - - df.count shouldEqual 4 - - val inputPathLarger: String = prepareFiles((1 to 300).map(_ => path).toArray, minioClient) - - val dfLarger = sparkSessionMinio.read - .format("geopackage") - .option("tableName", "point1") - .load(inputPathLarger) - - dfLarger.count shouldEqual 300 * 4 - - container.stop() - } - } - - describe("_metadata hidden column support") { - it("should expose _metadata struct with all expected fields") { - val df = readFeatureData("point1") - val metaDf = df.select("_metadata") - val metaSchema = metaDf.schema.fields.head.dataType.asInstanceOf[StructType] - val fieldNames = metaSchema.fieldNames.toSet - fieldNames should contain("file_path") - fieldNames should contain("file_name") - fieldNames should contain("file_size") - fieldNames should contain("file_block_start") - fieldNames should contain("file_block_length") - fieldNames should contain("file_modification_time") - } - - it("should not include _metadata in select(*)") { - val df = readFeatureData("point1") - val starCols = df.select("*").columns.toSet - starCols should not contain "_metadata" - } - - it("should return correct file_path and file_name in _metadata") { - val df = readFeatureData("point1") - val row = df.select("_metadata.file_path", "_metadata.file_name").head() - val filePath = row.getString(0) - val fileName = row.getString(1) - filePath should endWith("example.gpkg") - fileName shouldEqual "example.gpkg" - } - - it("should return actual file_size matching the .gpkg file on disk") { - val df = readFeatureData("point1") - val metaFileSize = df.select("_metadata.file_size").head().getLong(0) - val actualFile = new java.io.File(path) - metaFileSize shouldEqual actualFile.length() - } - - it("should return file_block_start=0 and file_block_length=file_size") { - val df = readFeatureData("point1") - val row = df - .select( - "_metadata.file_block_start", - "_metadata.file_block_length", - "_metadata.file_size") - .head() - row.getLong(0) shouldEqual 0L - row.getLong(1) shouldEqual row.getLong(2) - } - - it("should return file_modification_time matching the .gpkg file on disk") { - val df = readFeatureData("point1") - val metaModTime = df.select("_metadata.file_modification_time").head().getTimestamp(0) - val actualFile = new java.io.File(path) - val expectedModTime = new java.sql.Timestamp(actualFile.lastModified()) - metaModTime shouldEqual expectedModTime - } - - it("should allow filtering on _metadata fields") { - val df = readFeatureData("point1") - val filtered = df.filter(df("_metadata.file_name") === "example.gpkg") - filtered.count() shouldEqual df.count() - val empty = df.filter(df("_metadata.file_name") === "nonexistent.gpkg") - empty.count() shouldEqual 0 - } - - it("should select _metadata along with data columns") { - val df = readFeatureData("point1") - val result = df.select("id", "_metadata.file_name").head() - result.getInt(0) shouldEqual 1 - result.getString(1) shouldEqual "example.gpkg" - } - } - - private def readFeatureData(tableName: String): DataFrame = { - sparkSession.read - .format("geopackage") - .option("tableName", tableName) - .load(path) - } - - private def prepareFiles(paths: Array[String], minioClient: MinioClient): String = { - val key = "geopackage" - - paths.foreach(path => { - val fis = new FileInputStream(path); - putFileIntoBucket( - "sedona", - s"${key}/${scala.util.Random.nextInt(1000000000)}.geopackage", - fis, - minioClient) - }) - - s"s3a://sedona/$key" - } - - private def prepareFile(name: String, path: String, minioClient: MinioClient): String = { - val fis = new FileInputStream(path); - putFileIntoBucket("sedona", name, fis, minioClient) - - s"s3a://sedona/$name" - } - - private val POINT_1 = "POINT (-104.801918 39.720014)" - private val POINT_2 = "POINT (-104.802987 39.717703)" - private val POINT_3 = "POINT (-104.807496 39.714085)" - private val POINT_4 = "POINT (-104.79948 39.714729)" - private val LINESTRING_1 = - "LINESTRING (-104.800614 39.720721, -104.802174 39.720726, -104.802584 39.72066, -104.803088 39.720477, -104.803474 39.720209)" - private val LINESTRING_2 = - "LINESTRING (-104.809612 39.718379, -104.806638 39.718372, -104.806236 39.718439, -104.805939 39.718536, -104.805654 39.718677, -104.803652 39.720095)" - private val LINESTRING_3 = - "LINESTRING (-104.806344 39.722425, -104.805854 39.722634, -104.805656 39.722647, -104.803749 39.722641, -104.803769 39.721849, -104.803806 39.721725, -104.804382 39.720865)" - private val POLYGON_1 = - "POLYGON ((-104.802246 39.720343, -104.802246 39.719753, -104.802183 39.719754, -104.802184 39.719719, -104.802138 39.719694, -104.802097 39.719691, -104.802096 39.719648, -104.801646 39.719648, -104.801644 39.719722, -104.80155 39.719723, -104.801549 39.720207, -104.801648 39.720207, -104.801648 39.720341, -104.802246 39.720343))" - private val POLYGON_2 = - "POLYGON ((-104.802259 39.719604, -104.80226 39.71955, -104.802281 39.719416, -104.802332 39.719372, -104.802081 39.71924, -104.802044 39.71929, -104.802027 39.719278, -104.802044 39.719229, -104.801785 39.719129, -104.801639 39.719413, -104.801649 39.719472, -104.801694 39.719524, -104.801753 39.71955, -104.80175 39.719606, -104.80194 39.719606, -104.801939 39.719555, -104.801977 39.719556, -104.801979 39.719606, -104.802259 39.719604), (-104.80213 39.71944, -104.802133 39.71949, -104.802148 39.71949, -104.80218 39.719473, -104.802187 39.719456, -104.802182 39.719439, -104.802088 39.719387, -104.802047 39.719427, -104.801858 39.719342, -104.801883 39.719294, -104.801832 39.719284, -104.801787 39.719298, -104.801763 39.719331, -104.801823 39.719352, -104.80179 39.71942, -104.801722 39.719404, -104.801715 39.719445, -104.801748 39.719484, -104.801809 39.719494, -104.801816 39.719439, -104.80213 39.71944))" - private val POLYGON_3 = - "POLYGON ((-104.802867 39.718122, -104.802369 39.717845, -104.802571 39.71763, -104.803066 39.717909, -104.802867 39.718122))" -}