You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@sedona.apache.org by "jiayuasu (via GitHub)" <gi...@apache.org> on 2026/02/24 09:31:59 UTC
[PR] [SEDONA-737] Add DataSourceV2-based raster data source for GeoTIFF loading with tiling on read [sedona]
jiayuasu opened a new pull request, #2673:
URL: https://github.com/apache/sedona/pull/2673
## Summary
Fix `toEpsgCode()` to work correctly with PROJJSON input and add `toAuthority()` for non-EPSG CRS identification.
When a CRS is constructed from PROJJSON (e.g., from pyproj output), `toEpsgCode()` previously failed to identify well-known codes because:
1. The `id` field's authority/code was not preserved during parsing
2. Datum names from PROJJSON were not stored or used for identification
3. Projection method names like `"Transverse Mercator"` were not normalized to PROJ short names (`"tmerc"`)
4. Ellipsoid-only comparison could not distinguish NAD83 (GRS 1980) from WGS 84
## Changes
### New API: `toAuthority()`
- `Proj.toAuthority()` / `CRSSerializer.toAuthority()` returns `{"authority", "code"}` (e.g., `{"EPSG", "4326"}` or `{"IAU", "49900"}`)
- Similar to pyproj's `CRS.to_authority()` — supports non-EPSG authorities
- `toEpsgCode()` now delegates to `toAuthority()` and filters for EPSG-only results
### Three-phase CRS identification (mirrors pyproj)
1. **srsCode** — if the PROJJSON `id` field was parsed, return it directly
2. **Datum name** — look up well-known datum names (restricted to geographic CRS)
3. **Parameter matching** — compare projection parameters against known EPSG definitions
### Parser fixes (ProjJsonTransformer)
- Store `datum.name` / `datum_ensemble.name` in `datumCode` field
- Store `id.authority:id.code` in `srsCode` field (with Gson Double→int handling)
- Preserve `srsCode` on `Proj` construction from `"EPSG:4326"` strings
### Robust matching (CRSSerializer)
- `WKT_TO_PROJ_METHOD` — reverse map with all 43 EPSG canonical PROJJSON method names
- `DATUM_NAME_TO_EPSG` — ~27 common datums sourced from the PROJ database (EPSG registry)
- `datumCodesMatch()` — normalizes both sides to EPSG codes for comparison; rejects unknown datums (matching pyproj behavior)
- `matchesDefinition()` — now checks inverse flattening (`rf`) to distinguish GRS 1980 from WGS 84
## pyproj behavior comparison
The following table shows behavior alignment with pyproj 3.7.2 (PROJ 9.6.0):
| Input | pyproj `to_epsg()` | proj4sedona `toEpsgCode()` (before) | proj4sedona `toEpsgCode()` (after) |
|---|---|---|---|
| PROJJSON WGS 84 with `id.code=4326` | 4326 | null | EPSG:4326 |
| PROJJSON NAD83(2011) with `id.code=6318` | 6318 | null | EPSG:6318 |
| PROJJSON UTM 32N with `id.code=32632` | 32632 | null | EPSG:32632 |
| PROJJSON IAU:49900 | None (not EPSG) | null | null |
| PROJJSON NAD83(2011) **no id**, datum name present | 6318 | null | EPSG:6318 |
| PROJJSON unknown datum + GRS 1980, **no id** | None (confidence 60%) | EPSG:4269 (false positive) | null |
| PROJJSON unknown datum + WGS 84, **no id** | None (confidence 60%) | EPSG:4326 (false positive) | null |
| `"EPSG:4326"` string | 4326 | EPSG:4326 | EPSG:4326 |
| `"+proj=longlat +datum=WGS84"` | 4326 | EPSG:4326 | EPSG:4326 |
| `"EPSG:3857"` string | 3857 | EPSG:3857 | EPSG:3857 |
## Performance
Microbenchmark comparing old vs new `matchesDefinition()` (10,000 iterations, Apple M1 Max):
| Scenario | Old (equalsIgnoreCase) | New (datumCodesMatch) | Delta |
|---|---|---|---|
| WGS 84 match | 0.4 µs | 0.4 µs | noise |
| NAD83 match | 0.4 µs | 0.4 µs | noise |
| Unknown datum reject | 0.3 µs | 0.3 µs | noise |
The datum normalization lookup adds negligible overhead because it's dominated by the `new Proj(code)` call inside `matchesDefinition()`. Full test suite time is unchanged (~10s).
## Tests
- 13 new tests covering PROJJSON identification, `toAuthority()`, datum name lookup, unknown datum rejection, IAU authority, and round-trips
- All 661 tests passing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "jiayuasu (via GitHub)" <gi...@apache.org>.
jiayuasu commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2851273580
##########
common/src/main/java/org/apache/sedona/common/raster/TileGenerator.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.common.raster;
+
+import java.awt.Rectangle;
+import java.awt.image.Raster;
+import java.awt.image.RenderedImage;
+import java.awt.image.WritableRaster;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import javax.media.jai.RasterFactory;
+import org.apache.sedona.common.utils.ImageUtils;
+import org.apache.sedona.common.utils.RasterUtils;
+import org.geotools.api.metadata.spatial.PixelOrientation;
+import org.geotools.api.referencing.datum.PixelInCell;
+import org.geotools.coverage.GridSampleDimension;
+import org.geotools.coverage.grid.GridCoverage2D;
+import org.geotools.coverage.grid.GridEnvelope2D;
+import org.geotools.coverage.grid.GridGeometry2D;
+import org.geotools.referencing.operation.transform.AffineTransform2D;
+
+public class TileGenerator {
+
+ public static class Tile {
+
+ private final int tileX;
+ private final int tileY;
+ private final GridCoverage2D coverage;
+
+ public Tile(int tileX, int tileY, GridCoverage2D coverage) {
+ this.tileX = tileX;
+ this.tileY = tileY;
+ this.coverage = coverage;
+ }
+
+ public int getTileX() {
+ return tileX;
+ }
+
+ public int getTileY() {
+ return tileY;
+ }
+
+ public GridCoverage2D getCoverage() {
+ return coverage;
+ }
+ }
+
+ /**
+ * Generate tiles from an in-db grid coverage. The generated tiles are also in-db grid coverages.
+ * Pixel data will be copied into the tiles one tile at a time via a lazy iterator.
+ *
+ * @param gridCoverage2D the in-db grid coverage
+ * @param bandIndices the indices of the bands to select (1-based)
+ * @param tileWidth the width of the tiles
+ * @param tileHeight the height of the tiles
+ * @param padWithNoData whether to pad the tiles with no data value
+ * @param padNoDataValue the no data value for padded tiles, only used when padWithNoData is true.
+ * If the value is NaN, the no data value of the original band will be used.
+ * @return a lazy iterator of tiles
+ */
+ public static InDbTileIterator generateInDbTiles(
+ GridCoverage2D gridCoverage2D,
+ int[] bandIndices,
+ int tileWidth,
+ int tileHeight,
+ boolean padWithNoData,
+ double padNoDataValue) {
+ return new InDbTileIterator(
+ gridCoverage2D, bandIndices, tileWidth, tileHeight, padWithNoData, padNoDataValue);
+ }
+
+ public abstract static class TileIterator implements Iterator<Tile> {
+ protected final GridCoverage2D gridCoverage2D;
+ protected int numTileX;
+ protected int numTileY;
+ protected int tileX;
+ protected int tileY;
+ protected boolean autoDisposeSource = false;
+ protected Runnable disposeFunction = null;
+
+ TileIterator(GridCoverage2D gridCoverage2D) {
+ this.gridCoverage2D = gridCoverage2D;
+ }
+
+ protected void initialize(int numTileX, int numTileY) {
+ this.numTileX = numTileX;
+ this.numTileY = numTileY;
+ this.tileX = 0;
+ this.tileY = 0;
+ }
+
+ /**
+ * Set whether to dispose the grid coverage when the iterator reaches the end. Default is false.
+ *
+ * @param autoDisposeSource whether to dispose the grid coverage
+ */
+ public void setAutoDisposeSource(boolean autoDisposeSource) {
+ this.autoDisposeSource = autoDisposeSource;
+ }
+
+ public void setDisposeFunction(Runnable disposeFunction) {
+ this.disposeFunction = disposeFunction;
+ }
+
+ public int getNumTileX() {
+ return numTileX;
+ }
+
+ public int getNumTileY() {
+ return numTileY;
+ }
+
+ public int getNumTiles() {
+ return numTileX * numTileY;
+ }
+
+ protected abstract Tile generateTile();
+
+ @Override
+ public boolean hasNext() {
+ if (numTileX == 0 || numTileY == 0) {
+ return false;
+ }
+ return tileY < numTileY;
+ }
+
+ @Override
+ public Tile next() {
+ // Check if current tile coordinate is valid
+ if (tileX >= numTileX || tileY >= numTileY) {
+ throw new NoSuchElementException();
+ }
+
+ Tile tile = generateTile();
+
+ // Advance to the next tile
+ tileX += 1;
+ if (tileX >= numTileX) {
+ tileX = 0;
+ tileY += 1;
+
+ // Dispose the grid coverage if we are at the end
+ if (tileY >= numTileY) {
+ if (autoDisposeSource) {
+ gridCoverage2D.dispose(true);
+ }
+ if (disposeFunction != null) {
+ disposeFunction.run();
+ }
+ }
+ }
+
+ return tile;
+ }
+ }
+
+ public static class InDbTileIterator extends TileIterator {
+ private final int[] bandIndices;
+ private final int tileWidth;
+ private final int tileHeight;
+ private final boolean padWithNoData;
+ private final double padNoDataValue;
+ private final AffineTransform2D affine;
+ private final RenderedImage image;
+ private final double[] noDataValues;
+ private final int imageWidth;
+ private final int imageHeight;
+
+ public InDbTileIterator(
+ GridCoverage2D gridCoverage2D,
+ int[] bandIndices,
+ int tileWidth,
+ int tileHeight,
+ boolean padWithNoData,
+ double padNoDataValue) {
+ super(gridCoverage2D);
+ this.bandIndices = bandIndices;
+ this.tileWidth = tileWidth;
+ this.tileHeight = tileHeight;
+ this.padWithNoData = padWithNoData;
+ this.padNoDataValue = padNoDataValue;
+
+ affine = RasterUtils.getAffineTransform(gridCoverage2D, PixelOrientation.CENTER);
+ image = gridCoverage2D.getRenderedImage();
+ noDataValues = new double[bandIndices.length];
+ for (int i = 0; i < bandIndices.length; i++) {
+ noDataValues[i] =
+ RasterUtils.getNoDataValue(gridCoverage2D.getSampleDimension(bandIndices[i] - 1));
+ }
+ imageWidth = image.getWidth();
+ imageHeight = image.getHeight();
+ int numTileX = (int) Math.ceil((double) imageWidth / tileWidth);
+ int numTileY = (int) Math.ceil((double) imageHeight / tileHeight);
+ initialize(numTileX, numTileY);
+ }
+
+ @Override
+ protected Tile generateTile() {
+ // Process the current tile
+ int x0 = tileX * tileWidth + image.getMinX();
+ int y0 = tileY * tileHeight + image.getMinY();
+
+ // Rect to copy from the original image
+ int rectWidth = Math.min(tileWidth, imageWidth - x0);
+ int rectHeight = Math.min(tileHeight, imageHeight - y0);
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "jiayuasu (via GitHub)" <gi...@apache.org>.
jiayuasu commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2851396977
##########
spark/common/src/test/scala/org/apache/sedona/sql/rasterIOTest.scala:
##########
@@ -239,5 +248,283 @@ class rasterIOTest extends TestBaseScala with BeforeAndAfter with GivenWhenThen
}
}
- override def afterAll(): Unit = FileUtils.deleteDirectory(new File(tempDir))
+ describe("Raster read test") {
+ it("should read geotiff using raster source with explicit tiling") {
+ val rasterDf = sparkSession.read
+ .format("raster")
+ .options(Map("retile" -> "true", "tileWidth" -> "64"))
+ .load(rasterdatalocation)
+ assert(rasterDf.count() > 100)
+ rasterDf.collect().foreach { row =>
+ val raster = row.getAs[Object](0).asInstanceOf[GridCoverage2D]
+ assert(raster.getGridGeometry.getGridRange2D.width <= 64)
+ assert(raster.getGridGeometry.getGridRange2D.height <= 64)
+ val x = row.getInt(1)
+ val y = row.getInt(2)
+ assert(x >= 0 && y >= 0)
+ raster.dispose(true)
+ }
+
+ // Test projection push-down
+ rasterDf.selectExpr("y", "rast as r").collect().foreach { row =>
+ val raster = row.getAs[Object](1).asInstanceOf[GridCoverage2D]
+ assert(raster.getGridGeometry.getGridRange2D.width <= 64)
+ assert(raster.getGridGeometry.getGridRange2D.height <= 64)
+ val y = row.getInt(0)
+ assert(y >= 0)
+ raster.dispose(true)
+ }
+ }
+
+ it("should tile geotiff using raster source with padding enabled") {
+ val rasterDf = sparkSession.read
+ .format("raster")
+ .options(Map("retile" -> "true", "tileWidth" -> "64", "padWithNoData" -> "true"))
+ .load(rasterdatalocation)
+ assert(rasterDf.count() > 100)
+ rasterDf.collect().foreach { row =>
+ val raster = row.getAs[Object](0).asInstanceOf[GridCoverage2D]
+ assert(raster.getGridGeometry.getGridRange2D.width == 64)
+ assert(raster.getGridGeometry.getGridRange2D.height == 64)
+ val x = row.getInt(1)
+ val y = row.getInt(2)
+ assert(x >= 0 && y >= 0)
+ raster.dispose(true)
+ }
+ }
+
+ it("should push down limit and sample to data source") {
+ FileUtils.cleanDirectory(new File(tempDir))
+
+ val sourceDir = new File(rasterdatalocation)
+ val files = sourceDir.listFiles().filter(_.isFile)
+ var numUniqueFiles = 0
+ var numTotalFiles = 0
+ files.foreach { file =>
+ if (file.getPath.endsWith(".tif") || file.getPath.endsWith(".tiff")) {
+ // Create 4 copies for each file
+ for (i <- 0 until 4) {
+ val destFile = new File(tempDir + "/" + file.getName + "_" + i)
+ FileUtils.copyFile(file, destFile)
+ numTotalFiles += 1
+ }
+ numUniqueFiles += 1
+ }
+ }
+
+ val df = sparkSession.read
+ .format("raster")
+ .options(Map("retile" -> "false"))
+ .load(tempDir)
+ .withColumn("width", expr("RS_Width(rast)"))
+
+ val dfWithLimit = df.limit(numUniqueFiles)
+ val plan = queryPlan(dfWithLimit)
+ // Global/local limits are all pushed down to data source
+ assert(plan.collect { case e: LimitExec => e }.isEmpty)
+ assert(dfWithLimit.count() == numUniqueFiles)
+
+ val dfWithSample = df.sample(0.3, seed = 42)
+ val planSample = queryPlan(dfWithSample)
+ // Sample is pushed down to data source
+ assert(planSample.collect { case e: SampleExec => e }.isEmpty)
+ val count = dfWithSample.count()
+ assert(count >= numTotalFiles * 0.1 && count <= numTotalFiles * 0.5)
+
+ val dfWithSampleAndLimit = df.sample(0.5, seed = 42).limit(numUniqueFiles)
+ val planBoth = queryPlan(dfWithSampleAndLimit)
+ assert(planBoth.collect { case e: LimitExec => e }.isEmpty)
+ assert(planBoth.collect { case e: SampleExec => e }.isEmpty)
+ assert(dfWithSampleAndLimit.count() == numUniqueFiles)
+
+ // Limit and sample cannot be fully pushed down when retile is enabled
+ val dfReTiledWithSampleAndLimit = sparkSession.read
+ .format("raster")
+ .options(Map("retile" -> "true"))
+ .load(tempDir)
+ .sample(0.5, seed = 42)
+ .limit(numUniqueFiles)
+ dfReTiledWithSampleAndLimit.explain(true)
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "jiayuasu (via GitHub)" <gi...@apache.org>.
jiayuasu commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2851395766
##########
common/src/test/java/org/apache/sedona/common/raster/inputstream/HadoopImageInputStreamTest.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.common.raster.inputstream;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.RandomAccessFile;
+import java.nio.file.Files;
+import java.util.Random;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PositionedReadable;
+import org.apache.hadoop.fs.Seekable;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class HadoopImageInputStreamTest {
+ @Rule public TemporaryFolder temp = new TemporaryFolder();
+
+ private static final int TEST_FILE_SIZE = 1000;
+ private final Random random = new Random();
+ private File testFile;
+
+ @Before
+ public void setup() throws IOException {
+ testFile = temp.newFile();
+ prepareTestData(testFile);
+ }
+
+ @Test
+ public void testReadSequentially() throws IOException {
+ Path path = new Path(testFile.getPath());
+ try (HadoopImageInputStream stream = new HadoopImageInputStream(path);
+ InputStream in = new BufferedInputStream(Files.newInputStream(testFile.toPath()))) {
+ byte[] bActual = new byte[8];
+ byte[] bExpected = new byte[bActual.length];
+ while (true) {
+ int len = random.nextInt(bActual.length + 1);
+ int lenActual = stream.read(bActual, 0, len);
+ int lenExpected = in.read(bExpected, 0, len);
+ Assert.assertEquals(lenExpected, lenActual);
+ if (lenActual < 0) {
+ break;
+ }
+ Assert.assertArrayEquals(bExpected, bActual);
+ }
+ }
+ }
+
+ @Test
+ public void testReadRandomly() throws IOException {
+ Path path = new Path(testFile.getPath());
+ try (HadoopImageInputStream stream = new HadoopImageInputStream(path);
+ RandomAccessFile raf = new RandomAccessFile(testFile, "r")) {
+ byte[] bActual = new byte[8];
+ byte[] bExpected = new byte[bActual.length];
+ for (int k = 0; k < 1000; k++) {
+ int offset = random.nextInt(TEST_FILE_SIZE + 1);
+ int len = random.nextInt(bActual.length + 1);
+ stream.seek(offset);
+ raf.seek(offset);
+ int lenActual = stream.read(bActual, 0, len);
+ int lenExpected = raf.read(bExpected, 0, len);
+ Assert.assertEquals(lenExpected, lenActual);
+ if (lenActual < 0) {
+ continue;
+ }
+ Assert.assertArrayEquals(bExpected, bActual);
+ }
+
+ // Test seek to EOF.
+ stream.seek(TEST_FILE_SIZE);
+ int len = stream.read(bActual, 0, bActual.length);
+ Assert.assertEquals(-1, len);
+ }
+ }
+
+ @Test
+ public void testFromUnstableStream() throws IOException {
+ Path path = new Path(testFile.getPath());
+ FileSystem fs = path.getFileSystem(new Configuration());
+ try (FSDataInputStream unstable = new UnstableFSDataInputStream(fs.open(path));
+ HadoopImageInputStream stream = new HadoopImageInputStream(unstable);
+ InputStream in = new BufferedInputStream(Files.newInputStream(testFile.toPath()))) {
+ byte[] bActual = new byte[8];
+ byte[] bExpected = new byte[bActual.length];
+ while (true) {
+ int len = random.nextInt(bActual.length);
+ int lenActual = stream.read(bActual, 0, len);
+ int lenExpected = in.read(bExpected, 0, len);
+ Assert.assertEquals(lenExpected, lenActual);
+ if (lenActual < 0) {
+ break;
+ }
+ Assert.assertArrayEquals(bExpected, bActual);
+ }
+ }
+ }
+
+ private void prepareTestData(File testFile) throws IOException {
+ try (OutputStream out = new BufferedOutputStream(Files.newOutputStream(testFile.toPath()))) {
+ for (int k = 0; k < TEST_FILE_SIZE; k++) {
+ out.write(random.nextInt());
+ }
+ }
+ }
+
+ /**
+ * An FSDataInputStream that sometimes return less data than requested when calling read(byte[],
+ * int, int).
+ */
+ private static class UnstableFSDataInputStream extends FSDataInputStream {
+ public UnstableFSDataInputStream(FSDataInputStream in) {
+ super(new UnstableInputStream(in.getWrappedStream()));
+ }
+
+ private static class UnstableInputStream extends InputStream
+ implements Seekable, PositionedReadable {
+ private final InputStream wrapped;
+ private final Random random = new Random();
+
+ UnstableInputStream(InputStream in) {
+ wrapped = in;
+ }
+
+ @Override
+ public void close() throws IOException {
+ wrapped.close();
+ }
+
+ @Override
+ public int read() throws IOException {
+ return wrapped.read();
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ // Make this read unstable, i.e. sometimes return less data than requested.
+ int unstableLen = random.nextInt(len + 1);
Review Comment:
Fixed
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/raster/RasterConstructors.scala:
##########
@@ -107,39 +109,35 @@ private[apache] case class RS_TileExplode(children: Seq[Expression])
override def eval(input: InternalRow): TraversableOnce[InternalRow] = {
val raster = arguments.rasterExpr.toRaster(input)
- try {
- val bandIndices = arguments.bandIndicesExpr.eval(input).asInstanceOf[ArrayData] match {
- case null => null
- case value: Any => value.toIntArray
- }
- val tileWidth = arguments.tileWidthExpr.eval(input).asInstanceOf[Int]
- val tileHeight = arguments.tileHeightExpr.eval(input).asInstanceOf[Int]
- val padWithNoDataValue = arguments.padWithNoDataExpr.eval(input).asInstanceOf[Boolean]
- val noDataValue = arguments.noDataValExpr.eval(input) match {
- case null => Double.NaN
- case value: Integer => value.toDouble
- case value: Decimal => value.toDouble
- case value: Float => value.toDouble
- case value: Double => value
- case value: Any =>
- throw new IllegalArgumentException(
- "Unsupported class for noDataValue: " + value.getClass)
- }
- val tiles = RasterConstructors.generateTiles(
- raster,
- bandIndices,
- tileWidth,
- tileHeight,
- padWithNoDataValue,
- noDataValue)
- tiles.map { tile =>
- val gridCoverage2D = tile.getCoverage
- val row = InternalRow(tile.getTileX, tile.getTileY, gridCoverage2D.serialize)
- gridCoverage2D.dispose(true)
- row
- }
- } finally {
- raster.dispose(true)
+ val bandIndices = arguments.bandIndicesExpr.eval(input).asInstanceOf[ArrayData] match {
+ case null => null
+ case value: Any => value.toIntArray
+ }
+ val tileWidth = arguments.tileWidthExpr.eval(input).asInstanceOf[Int]
+ val tileHeight = arguments.tileHeightExpr.eval(input).asInstanceOf[Int]
+ val padWithNoDataValue = arguments.padWithNoDataExpr.eval(input).asInstanceOf[Boolean]
+ val noDataValue = arguments.noDataValExpr.eval(input) match {
+ case null => Double.NaN
+ case value: Integer => value.toDouble
+ case value: Decimal => value.toDouble
+ case value: Float => value.toDouble
+ case value: Double => value
+ case value: Any =>
+ throw new IllegalArgumentException("Unsupported class for noDataValue: " + value.getClass)
+ }
+ val tileIterator = RasterConstructors.generateTiles(
+ raster,
+ bandIndices,
+ tileWidth,
+ tileHeight,
+ padWithNoDataValue,
+ noDataValue)
+ tileIterator.setAutoDisposeSource(true)
+ tileIterator.asScala.map { tile =>
+ val gridCoverage2D = tile.getCoverage
+ val row = InternalRow(tile.getTileX, tile.getTileY, gridCoverage2D.serialize)
+ gridCoverage2D.dispose(true)
+ row
}
Review Comment:
Fixed
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterPartitionReader.scala:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.sedona.common.raster.RasterConstructors
+import org.apache.sedona.common.raster.inputstream.HadoopImageInputStream
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter
+import org.apache.spark.sql.connector.read.PartitionReader
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.sedona_sql.UDT.RasterUDT
+import org.apache.spark.sql.sedona_sql.io.raster.RasterPartitionReader.rasterToInternalRows
+import org.apache.spark.sql.sedona_sql.io.raster.RasterTable.{MAX_AUTO_TILE_SIZE, RASTER, RASTER_NAME, TILE_X, TILE_Y}
+import org.apache.spark.sql.types.StructType
+import org.geotools.coverage.grid.GridCoverage2D
+
+import java.net.URI
+import scala.collection.JavaConverters._
+
+class RasterPartitionReader(
+ configuration: Configuration,
+ partitionedFiles: Array[PartitionedFile],
+ dataSchema: StructType,
+ rasterOptions: RasterOptions)
+ extends PartitionReader[InternalRow] {
+
+ // Track the current file index we're processing
+ private var currentFileIndex = 0
+
+ // Current raster being processed
+ private var currentRaster: GridCoverage2D = _
+
+ // Current image input stream (must be kept open while the raster is in use)
+ private var currentImageStream: HadoopImageInputStream = _
+
+ // Current row
+ private var currentRow: InternalRow = _
+
+ // Iterator for the current file's tiles
+ private var currentIterator: Iterator[InternalRow] = Iterator.empty
+
+ override def next(): Boolean = {
+ // If current iterator has more elements, return true
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+
+ // If current iterator is exhausted, but we have more files, load the next file
+ if (currentFileIndex < partitionedFiles.length) {
+ loadNextFile()
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+ }
+
+ // No more data
+ false
+ }
+
+ override def get(): InternalRow = {
+ currentRow
+ }
+
+ override def close(): Unit = {
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+ }
+
+ private def loadNextFile(): Unit = {
+ // Clean up previous raster and stream if exists
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+
+ if (currentFileIndex >= partitionedFiles.length) {
+ currentIterator = Iterator.empty
+ return
+ }
+
+ val partition = partitionedFiles(currentFileIndex)
+ val path = new Path(new URI(partition.filePath.toString()))
+
+ try {
+ // Open a stream-based reader instead of materializing the entire file as byte[].
+ // This avoids the 2 GB byte[] limit and reduces memory pressure for large files.
+ currentImageStream = new HadoopImageInputStream(path, configuration)
+
+ // Create in-db GridCoverage2D from GeoTiff stream. The RenderedImage is lazy -
+ // pixel data will only be decoded when accessed via image.getData(Rectangle).
+ currentRaster = RasterConstructors.fromGeoTiff(currentImageStream)
+ currentIterator = rasterToInternalRows(currentRaster, dataSchema, rasterOptions, path)
+ currentFileIndex += 1
+ } catch {
+ case e: Exception =>
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+ throw e
+ }
+ }
+}
+
+object RasterPartitionReader {
+ def rasterToInternalRows(
+ currentRaster: GridCoverage2D,
+ dataSchema: StructType,
+ rasterOptions: RasterOptions,
+ path: Path): Iterator[InternalRow] = {
+ val retile = rasterOptions.retile
+ val tileWidth = rasterOptions.tileWidth
+ val tileHeight = rasterOptions.tileHeight
+ val padWithNoData = rasterOptions.padWithNoData
+
+ val writer = new UnsafeRowWriter(dataSchema.length)
+ writer.resetRowWriter()
+
+ // Extract the file name from the path
+ val fileName = path.getName
+
+ if (retile) {
+ val (tw, th) = (tileWidth, tileHeight) match {
+ case (Some(tw), Some(th)) => (tw, th)
+ case (None, None) =>
+ // Use the internal tile size of the input raster
+ val tw = currentRaster.getRenderedImage.getTileWidth
+ val th = currentRaster.getRenderedImage.getTileHeight
+ val tileSizeError = {
+ """To resolve this issue, you can try one of the following methods:
+ | 1. Disable retile by setting `.option("retile", "false")`.
+ | 2. Explicitly set `tileWidth` and `tileHeight`.
+ | 3. Convert the raster to a Cloud Optimized GeoTIFF (COG) using tools like `gdal_translate`.
+ |""".stripMargin
+ }
+ if (tw >= MAX_AUTO_TILE_SIZE || th >= MAX_AUTO_TILE_SIZE) {
+ throw new IllegalArgumentException(
+ s"Internal tile size of $path is too large ($tw x $th). " + tileSizeError)
+ }
+ if (tw == 0 || th == 0) {
+ throw new IllegalArgumentException(
+ s"Internal tile size of $path contains zero ($tw x $th). " + tileSizeError)
+ }
+ if (tw / th > 10 || th / tw > 10) {
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "jiayuasu (via GitHub)" <gi...@apache.org>.
jiayuasu merged PR #2673:
URL: https://github.com/apache/sedona/pull/2673
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "jiayuasu (via GitHub)" <gi...@apache.org>.
jiayuasu commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2851271844
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterTable.scala:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.hadoop.fs.FileStatus
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.connector.catalog.SupportsRead
+import org.apache.spark.sql.connector.catalog.SupportsWrite
+import org.apache.spark.sql.connector.catalog.TableCapability
+import org.apache.spark.sql.connector.read.ScanBuilder
+import org.apache.spark.sql.connector.write.LogicalWriteInfo
+import org.apache.spark.sql.connector.write.WriteBuilder
+import org.apache.spark.sql.execution.datasources.v2.FileTable
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.sedona_sql.UDT.RasterUDT
+import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.sql.types.StructField
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import java.util.{Set => JSet}
+
+case class RasterTable(
+ name: String,
+ sparkSession: SparkSession,
+ options: CaseInsensitiveStringMap,
+ paths: Seq[String],
+ userSpecifiedSchema: Option[StructType],
+ rasterOptions: RasterOptions,
+ fallbackFileFormat: Class[_ <: FileFormat])
+ extends FileTable(sparkSession, options, paths, userSpecifiedSchema)
+ with SupportsRead
+ with SupportsWrite {
+
+ override def inferSchema(files: Seq[FileStatus]): Option[StructType] =
+ Some(userSpecifiedSchema.getOrElse(RasterTable.inferSchema(rasterOptions)))
+
+ override def formatName: String = "Raster"
+
+ override def capabilities(): JSet[TableCapability] =
+ java.util.EnumSet.of(TableCapability.BATCH_READ)
+
+ override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = {
+ RasterScanBuilder(sparkSession, fileIndex, schema, dataSchema, options, rasterOptions)
+ }
+
+ override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder =
+ // Note: this code path will never be taken, since Spark will always fall back to V1
+ // data source when writing to File source v2. See SPARK-28396: File source v2 write
+ // path is currently broken.
+ null
Review Comment:
Not an issue
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/raster/RasterConstructors.scala:
##########
@@ -107,39 +109,35 @@ private[apache] case class RS_TileExplode(children: Seq[Expression])
override def eval(input: InternalRow): TraversableOnce[InternalRow] = {
val raster = arguments.rasterExpr.toRaster(input)
- try {
- val bandIndices = arguments.bandIndicesExpr.eval(input).asInstanceOf[ArrayData] match {
- case null => null
- case value: Any => value.toIntArray
- }
- val tileWidth = arguments.tileWidthExpr.eval(input).asInstanceOf[Int]
- val tileHeight = arguments.tileHeightExpr.eval(input).asInstanceOf[Int]
- val padWithNoDataValue = arguments.padWithNoDataExpr.eval(input).asInstanceOf[Boolean]
- val noDataValue = arguments.noDataValExpr.eval(input) match {
- case null => Double.NaN
- case value: Integer => value.toDouble
- case value: Decimal => value.toDouble
- case value: Float => value.toDouble
- case value: Double => value
- case value: Any =>
- throw new IllegalArgumentException(
- "Unsupported class for noDataValue: " + value.getClass)
- }
- val tiles = RasterConstructors.generateTiles(
- raster,
- bandIndices,
- tileWidth,
- tileHeight,
- padWithNoDataValue,
- noDataValue)
- tiles.map { tile =>
- val gridCoverage2D = tile.getCoverage
- val row = InternalRow(tile.getTileX, tile.getTileY, gridCoverage2D.serialize)
- gridCoverage2D.dispose(true)
- row
- }
- } finally {
- raster.dispose(true)
+ val bandIndices = arguments.bandIndicesExpr.eval(input).asInstanceOf[ArrayData] match {
+ case null => null
+ case value: Any => value.toIntArray
+ }
+ val tileWidth = arguments.tileWidthExpr.eval(input).asInstanceOf[Int]
+ val tileHeight = arguments.tileHeightExpr.eval(input).asInstanceOf[Int]
+ val padWithNoDataValue = arguments.padWithNoDataExpr.eval(input).asInstanceOf[Boolean]
+ val noDataValue = arguments.noDataValExpr.eval(input) match {
+ case null => Double.NaN
+ case value: Integer => value.toDouble
+ case value: Decimal => value.toDouble
+ case value: Float => value.toDouble
+ case value: Double => value
+ case value: Any =>
+ throw new IllegalArgumentException("Unsupported class for noDataValue: " + value.getClass)
+ }
+ val tileIterator = RasterConstructors.generateTiles(
+ raster,
+ bandIndices,
+ tileWidth,
+ tileHeight,
+ padWithNoDataValue,
+ noDataValue)
+ tileIterator.setAutoDisposeSource(true)
+ tileIterator.asScala.map { tile =>
+ val gridCoverage2D = tile.getCoverage
+ val row = InternalRow(tile.getTileX, tile.getTileY, gridCoverage2D.serialize)
+ gridCoverage2D.dispose(true)
+ row
Review Comment:
Not an issue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2851315423
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterOptions.scala:
##########
@@ -20,16 +20,52 @@ package org.apache.spark.sql.sedona_sql.io.raster
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
-private[io] class RasterOptions(@transient private val parameters: CaseInsensitiveMap[String])
+class RasterOptions(@transient private val parameters: CaseInsensitiveMap[String])
extends Serializable {
def this(parameters: Map[String, String]) = this(CaseInsensitiveMap(parameters))
+ // The following options are used to read raster data
+
+ /**
+ * Whether to retile the raster data. If true, the raster data will be retiled into smaller
+ * tiles. If false, the raster data will be read as a single tile.
+ */
+ val retile: Boolean = parameters.getOrElse("retile", "true").toBoolean
+
+ /**
+ * The width of the tile. This is only effective when retile is true. If retile is true and
+ * tileWidth is not set, the default value is the width of the internal tiles in the raster
+ * files. Each raster file may have different internal tile sizes.
+ */
+ val tileWidth: Option[Int] = parameters.get("tileWidth").map(_.toInt)
+
+ /**
+ * The height of the tile. This is only effective when retile is true. If retile is true and
+ * tileHeight is not set, the default value is the same as tileWidth. If tileHeight is set,
+ * tileWidth must be set as well.
+ */
+ val tileHeight: Option[Int] = parameters
+ .get("tileHeight")
+ .map { value =>
+ require(tileWidth.isDefined, "tileWidth must be set when tileHeight is set")
+ value.toInt
+ }
+ .orElse(tileWidth)
+
Review Comment:
`tileHeight` validation currently throws when `tileHeight` is set without `tileWidth` even if `retile=false` (where tiling options are documented as having no effect). Consider only enforcing the `tileWidth`/`tileHeight` pairing when `retile=true`, or ignoring tiling-specific options entirely when `retile=false` to avoid surprising failures.
##########
common/src/main/java/org/apache/sedona/common/raster/inputstream/HadoopImageInputStream.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.common.raster.inputstream;
+
+import java.io.IOException;
+import javax.imageio.stream.ImageInputStreamImpl;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+/** An ImageInputStream that reads image data from a Hadoop FileSystem. */
+public class HadoopImageInputStream extends ImageInputStreamImpl {
+
+ private final FSDataInputStream stream;
+ private final Path path;
+ private final Configuration conf;
+
+ public HadoopImageInputStream(Path path, Configuration conf) throws IOException {
+ FileSystem fs = path.getFileSystem(conf);
+ stream = fs.open(path);
+ this.path = path;
+ this.conf = conf;
+ }
+
+ public HadoopImageInputStream(Path path) throws IOException {
+ this(path, new Configuration());
+ }
+
+ public HadoopImageInputStream(FSDataInputStream stream) {
+ this.stream = stream;
+ this.path = null;
+ this.conf = null;
+ }
+
+ public Path getPath() {
+ return path;
+ }
+
+ public Configuration getConf() {
+ return conf;
+ }
+
+ @Override
+ public void close() throws IOException {
+ super.close();
+ stream.close();
+ }
+
+ @Override
+ public int read() throws IOException {
+ byte[] buf = new byte[1];
+ int ret_len = read(buf, 0, 1);
+ if (ret_len < 0) {
+ return ret_len;
+ }
+ return buf[0] & 0xFF;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ checkClosed();
+ bitOffset = 0;
+
+ if (len == 0) {
+ return 0;
+ }
+
+ // stream.read may return fewer data than requested, so we need to loop until we get all the
+ // data, or hit the end of the stream. We can not simply perform an incomplete read and return
+ // the number of bytes actually read, since the methods in ImageInputStreamImpl such as
+ // readInt() relies on this method and assumes that partial reads only happens when reading
+ // EOF. This might be a bug of imageio since they should invoke readFully() in such cases.
+ int remaining = len;
+ while (remaining > 0) {
+ int ret_len = stream.read(b, off, remaining);
+ if (ret_len < 0) {
+ // Hit EOF, no more data to read.
Review Comment:
`read(byte[], off, len)` loops until `remaining == 0`, but it never handles the case where `stream.read(...)` returns `0` for a positive `remaining`. That would result in an infinite loop / hang. Add a guard for `ret_len == 0` (e.g., retry with a bounded number of attempts or throw an `IOException` indicating a non-progressing stream).
##########
spark/common/src/test/scala/org/apache/sedona/sql/rasterIOTest.scala:
##########
@@ -239,5 +248,283 @@ class rasterIOTest extends TestBaseScala with BeforeAndAfter with GivenWhenThen
}
}
- override def afterAll(): Unit = FileUtils.deleteDirectory(new File(tempDir))
+ describe("Raster read test") {
+ it("should read geotiff using raster source with explicit tiling") {
+ val rasterDf = sparkSession.read
+ .format("raster")
+ .options(Map("retile" -> "true", "tileWidth" -> "64"))
+ .load(rasterdatalocation)
+ assert(rasterDf.count() > 100)
+ rasterDf.collect().foreach { row =>
+ val raster = row.getAs[Object](0).asInstanceOf[GridCoverage2D]
+ assert(raster.getGridGeometry.getGridRange2D.width <= 64)
+ assert(raster.getGridGeometry.getGridRange2D.height <= 64)
+ val x = row.getInt(1)
+ val y = row.getInt(2)
+ assert(x >= 0 && y >= 0)
+ raster.dispose(true)
+ }
+
+ // Test projection push-down
+ rasterDf.selectExpr("y", "rast as r").collect().foreach { row =>
+ val raster = row.getAs[Object](1).asInstanceOf[GridCoverage2D]
+ assert(raster.getGridGeometry.getGridRange2D.width <= 64)
+ assert(raster.getGridGeometry.getGridRange2D.height <= 64)
+ val y = row.getInt(0)
+ assert(y >= 0)
+ raster.dispose(true)
+ }
+ }
+
+ it("should tile geotiff using raster source with padding enabled") {
+ val rasterDf = sparkSession.read
+ .format("raster")
+ .options(Map("retile" -> "true", "tileWidth" -> "64", "padWithNoData" -> "true"))
+ .load(rasterdatalocation)
+ assert(rasterDf.count() > 100)
+ rasterDf.collect().foreach { row =>
+ val raster = row.getAs[Object](0).asInstanceOf[GridCoverage2D]
+ assert(raster.getGridGeometry.getGridRange2D.width == 64)
+ assert(raster.getGridGeometry.getGridRange2D.height == 64)
+ val x = row.getInt(1)
+ val y = row.getInt(2)
+ assert(x >= 0 && y >= 0)
+ raster.dispose(true)
+ }
+ }
+
+ it("should push down limit and sample to data source") {
+ FileUtils.cleanDirectory(new File(tempDir))
+
+ val sourceDir = new File(rasterdatalocation)
+ val files = sourceDir.listFiles().filter(_.isFile)
+ var numUniqueFiles = 0
+ var numTotalFiles = 0
+ files.foreach { file =>
+ if (file.getPath.endsWith(".tif") || file.getPath.endsWith(".tiff")) {
+ // Create 4 copies for each file
+ for (i <- 0 until 4) {
+ val destFile = new File(tempDir + "/" + file.getName + "_" + i)
+ FileUtils.copyFile(file, destFile)
+ numTotalFiles += 1
+ }
+ numUniqueFiles += 1
+ }
+ }
+
+ val df = sparkSession.read
+ .format("raster")
+ .options(Map("retile" -> "false"))
+ .load(tempDir)
+ .withColumn("width", expr("RS_Width(rast)"))
+
+ val dfWithLimit = df.limit(numUniqueFiles)
+ val plan = queryPlan(dfWithLimit)
+ // Global/local limits are all pushed down to data source
+ assert(plan.collect { case e: LimitExec => e }.isEmpty)
+ assert(dfWithLimit.count() == numUniqueFiles)
+
+ val dfWithSample = df.sample(0.3, seed = 42)
+ val planSample = queryPlan(dfWithSample)
+ // Sample is pushed down to data source
+ assert(planSample.collect { case e: SampleExec => e }.isEmpty)
+ val count = dfWithSample.count()
+ assert(count >= numTotalFiles * 0.1 && count <= numTotalFiles * 0.5)
+
+ val dfWithSampleAndLimit = df.sample(0.5, seed = 42).limit(numUniqueFiles)
+ val planBoth = queryPlan(dfWithSampleAndLimit)
+ assert(planBoth.collect { case e: LimitExec => e }.isEmpty)
+ assert(planBoth.collect { case e: SampleExec => e }.isEmpty)
+ assert(dfWithSampleAndLimit.count() == numUniqueFiles)
+
+ // Limit and sample cannot be fully pushed down when retile is enabled
+ val dfReTiledWithSampleAndLimit = sparkSession.read
+ .format("raster")
+ .options(Map("retile" -> "true"))
+ .load(tempDir)
+ .sample(0.5, seed = 42)
+ .limit(numUniqueFiles)
+ dfReTiledWithSampleAndLimit.explain(true)
Review Comment:
`explain(true)` in tests prints verbose plans to stdout/stderr and can create noisy CI logs. Consider removing it (the plan is already inspected via `queryPlan(...)`).
```suggestion
```
##########
common/src/test/java/org/apache/sedona/common/raster/inputstream/HadoopImageInputStreamTest.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.common.raster.inputstream;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.RandomAccessFile;
+import java.nio.file.Files;
+import java.util.Random;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PositionedReadable;
+import org.apache.hadoop.fs.Seekable;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class HadoopImageInputStreamTest {
+ @Rule public TemporaryFolder temp = new TemporaryFolder();
+
+ private static final int TEST_FILE_SIZE = 1000;
+ private final Random random = new Random();
+ private File testFile;
+
+ @Before
+ public void setup() throws IOException {
+ testFile = temp.newFile();
+ prepareTestData(testFile);
+ }
+
+ @Test
+ public void testReadSequentially() throws IOException {
+ Path path = new Path(testFile.getPath());
+ try (HadoopImageInputStream stream = new HadoopImageInputStream(path);
+ InputStream in = new BufferedInputStream(Files.newInputStream(testFile.toPath()))) {
+ byte[] bActual = new byte[8];
+ byte[] bExpected = new byte[bActual.length];
+ while (true) {
+ int len = random.nextInt(bActual.length + 1);
+ int lenActual = stream.read(bActual, 0, len);
+ int lenExpected = in.read(bExpected, 0, len);
+ Assert.assertEquals(lenExpected, lenActual);
+ if (lenActual < 0) {
+ break;
+ }
+ Assert.assertArrayEquals(bExpected, bActual);
+ }
+ }
+ }
+
+ @Test
+ public void testReadRandomly() throws IOException {
+ Path path = new Path(testFile.getPath());
+ try (HadoopImageInputStream stream = new HadoopImageInputStream(path);
+ RandomAccessFile raf = new RandomAccessFile(testFile, "r")) {
+ byte[] bActual = new byte[8];
+ byte[] bExpected = new byte[bActual.length];
+ for (int k = 0; k < 1000; k++) {
+ int offset = random.nextInt(TEST_FILE_SIZE + 1);
+ int len = random.nextInt(bActual.length + 1);
+ stream.seek(offset);
+ raf.seek(offset);
+ int lenActual = stream.read(bActual, 0, len);
+ int lenExpected = raf.read(bExpected, 0, len);
+ Assert.assertEquals(lenExpected, lenActual);
+ if (lenActual < 0) {
+ continue;
+ }
+ Assert.assertArrayEquals(bExpected, bActual);
+ }
+
+ // Test seek to EOF.
+ stream.seek(TEST_FILE_SIZE);
+ int len = stream.read(bActual, 0, bActual.length);
+ Assert.assertEquals(-1, len);
+ }
+ }
+
+ @Test
+ public void testFromUnstableStream() throws IOException {
+ Path path = new Path(testFile.getPath());
+ FileSystem fs = path.getFileSystem(new Configuration());
+ try (FSDataInputStream unstable = new UnstableFSDataInputStream(fs.open(path));
+ HadoopImageInputStream stream = new HadoopImageInputStream(unstable);
+ InputStream in = new BufferedInputStream(Files.newInputStream(testFile.toPath()))) {
+ byte[] bActual = new byte[8];
+ byte[] bExpected = new byte[bActual.length];
+ while (true) {
+ int len = random.nextInt(bActual.length);
+ int lenActual = stream.read(bActual, 0, len);
+ int lenExpected = in.read(bExpected, 0, len);
+ Assert.assertEquals(lenExpected, lenActual);
+ if (lenActual < 0) {
+ break;
+ }
+ Assert.assertArrayEquals(bExpected, bActual);
+ }
+ }
+ }
+
+ private void prepareTestData(File testFile) throws IOException {
+ try (OutputStream out = new BufferedOutputStream(Files.newOutputStream(testFile.toPath()))) {
+ for (int k = 0; k < TEST_FILE_SIZE; k++) {
+ out.write(random.nextInt());
+ }
+ }
+ }
+
+ /**
+ * An FSDataInputStream that sometimes return less data than requested when calling read(byte[],
+ * int, int).
+ */
+ private static class UnstableFSDataInputStream extends FSDataInputStream {
+ public UnstableFSDataInputStream(FSDataInputStream in) {
+ super(new UnstableInputStream(in.getWrappedStream()));
+ }
+
+ private static class UnstableInputStream extends InputStream
+ implements Seekable, PositionedReadable {
+ private final InputStream wrapped;
+ private final Random random = new Random();
+
+ UnstableInputStream(InputStream in) {
+ wrapped = in;
+ }
+
+ @Override
+ public void close() throws IOException {
+ wrapped.close();
+ }
+
+ @Override
+ public int read() throws IOException {
+ return wrapped.read();
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ // Make this read unstable, i.e. sometimes return less data than requested.
+ int unstableLen = random.nextInt(len + 1);
Review Comment:
`UnstableInputStream.read(byte[], off, len)` can choose `unstableLen = 0` when `len > 0`, which violates `InputStream`'s contract (must block until at least 1 byte or EOF) and will cause `HadoopImageInputStream.read(...)` to spin/hang. Ensure `unstableLen` is at least 1 when `len > 0` so this test reliably simulates partial (but progressing) reads.
```suggestion
// Make this read unstable, i.e. sometimes return less data than requested,
// but still obey InputStream's contract by not returning 0 when len > 0.
if (len == 0) {
return 0;
}
int unstableLen = 1 + random.nextInt(len);
```
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/raster/RasterConstructors.scala:
##########
@@ -107,39 +109,35 @@ private[apache] case class RS_TileExplode(children: Seq[Expression])
override def eval(input: InternalRow): TraversableOnce[InternalRow] = {
val raster = arguments.rasterExpr.toRaster(input)
- try {
- val bandIndices = arguments.bandIndicesExpr.eval(input).asInstanceOf[ArrayData] match {
- case null => null
- case value: Any => value.toIntArray
- }
- val tileWidth = arguments.tileWidthExpr.eval(input).asInstanceOf[Int]
- val tileHeight = arguments.tileHeightExpr.eval(input).asInstanceOf[Int]
- val padWithNoDataValue = arguments.padWithNoDataExpr.eval(input).asInstanceOf[Boolean]
- val noDataValue = arguments.noDataValExpr.eval(input) match {
- case null => Double.NaN
- case value: Integer => value.toDouble
- case value: Decimal => value.toDouble
- case value: Float => value.toDouble
- case value: Double => value
- case value: Any =>
- throw new IllegalArgumentException(
- "Unsupported class for noDataValue: " + value.getClass)
- }
- val tiles = RasterConstructors.generateTiles(
- raster,
- bandIndices,
- tileWidth,
- tileHeight,
- padWithNoDataValue,
- noDataValue)
- tiles.map { tile =>
- val gridCoverage2D = tile.getCoverage
- val row = InternalRow(tile.getTileX, tile.getTileY, gridCoverage2D.serialize)
- gridCoverage2D.dispose(true)
- row
- }
- } finally {
- raster.dispose(true)
+ val bandIndices = arguments.bandIndicesExpr.eval(input).asInstanceOf[ArrayData] match {
+ case null => null
+ case value: Any => value.toIntArray
+ }
+ val tileWidth = arguments.tileWidthExpr.eval(input).asInstanceOf[Int]
+ val tileHeight = arguments.tileHeightExpr.eval(input).asInstanceOf[Int]
+ val padWithNoDataValue = arguments.padWithNoDataExpr.eval(input).asInstanceOf[Boolean]
+ val noDataValue = arguments.noDataValExpr.eval(input) match {
+ case null => Double.NaN
+ case value: Integer => value.toDouble
+ case value: Decimal => value.toDouble
+ case value: Float => value.toDouble
+ case value: Double => value
+ case value: Any =>
+ throw new IllegalArgumentException("Unsupported class for noDataValue: " + value.getClass)
+ }
+ val tileIterator = RasterConstructors.generateTiles(
+ raster,
+ bandIndices,
+ tileWidth,
+ tileHeight,
+ padWithNoDataValue,
+ noDataValue)
+ tileIterator.setAutoDisposeSource(true)
+ tileIterator.asScala.map { tile =>
+ val gridCoverage2D = tile.getCoverage
+ val row = InternalRow(tile.getTileX, tile.getTileY, gridCoverage2D.serialize)
+ gridCoverage2D.dispose(true)
+ row
}
Review Comment:
`toRaster` can return `null` (e.g., when the input column is null). `RS_TileExplode.eval` currently passes that null into `RasterConstructors.generateTiles`, which will NPE. Add an explicit null check and return `Nil`/`Iterator.empty` (consistent with other raster expressions) when `raster` is null.
```suggestion
if (raster == null) {
Iterator.empty
} else {
val bandIndices = arguments.bandIndicesExpr.eval(input).asInstanceOf[ArrayData] match {
case null => null
case value: Any => value.toIntArray
}
val tileWidth = arguments.tileWidthExpr.eval(input).asInstanceOf[Int]
val tileHeight = arguments.tileHeightExpr.eval(input).asInstanceOf[Int]
val padWithNoDataValue = arguments.padWithNoDataExpr.eval(input).asInstanceOf[Boolean]
val noDataValue = arguments.noDataValExpr.eval(input) match {
case null => Double.NaN
case value: Integer => value.toDouble
case value: Decimal => value.toDouble
case value: Float => value.toDouble
case value: Double => value
case value: Any =>
throw new IllegalArgumentException("Unsupported class for noDataValue: " + value.getClass)
}
val tileIterator = RasterConstructors.generateTiles(
raster,
bandIndices,
tileWidth,
tileHeight,
padWithNoDataValue,
noDataValue)
tileIterator.setAutoDisposeSource(true)
tileIterator.asScala.map { tile =>
val gridCoverage2D = tile.getCoverage
val row = InternalRow(tile.getTileX, tile.getTileY, gridCoverage2D.serialize)
gridCoverage2D.dispose(true)
row
}
}
```
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterPartitionReader.scala:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.sedona.common.raster.RasterConstructors
+import org.apache.sedona.common.raster.inputstream.HadoopImageInputStream
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter
+import org.apache.spark.sql.connector.read.PartitionReader
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.sedona_sql.UDT.RasterUDT
+import org.apache.spark.sql.sedona_sql.io.raster.RasterPartitionReader.rasterToInternalRows
+import org.apache.spark.sql.sedona_sql.io.raster.RasterTable.{MAX_AUTO_TILE_SIZE, RASTER, RASTER_NAME, TILE_X, TILE_Y}
+import org.apache.spark.sql.types.StructType
+import org.geotools.coverage.grid.GridCoverage2D
+
+import java.net.URI
+import scala.collection.JavaConverters._
+
+class RasterPartitionReader(
+ configuration: Configuration,
+ partitionedFiles: Array[PartitionedFile],
+ dataSchema: StructType,
+ rasterOptions: RasterOptions)
+ extends PartitionReader[InternalRow] {
+
+ // Track the current file index we're processing
+ private var currentFileIndex = 0
+
+ // Current raster being processed
+ private var currentRaster: GridCoverage2D = _
+
+ // Current image input stream (must be kept open while the raster is in use)
+ private var currentImageStream: HadoopImageInputStream = _
+
+ // Current row
+ private var currentRow: InternalRow = _
+
+ // Iterator for the current file's tiles
+ private var currentIterator: Iterator[InternalRow] = Iterator.empty
+
+ override def next(): Boolean = {
+ // If current iterator has more elements, return true
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+
+ // If current iterator is exhausted, but we have more files, load the next file
+ if (currentFileIndex < partitionedFiles.length) {
+ loadNextFile()
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+ }
+
+ // No more data
+ false
+ }
+
+ override def get(): InternalRow = {
+ currentRow
+ }
+
+ override def close(): Unit = {
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+ }
+
+ private def loadNextFile(): Unit = {
+ // Clean up previous raster and stream if exists
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+
+ if (currentFileIndex >= partitionedFiles.length) {
+ currentIterator = Iterator.empty
+ return
+ }
+
+ val partition = partitionedFiles(currentFileIndex)
+ val path = new Path(new URI(partition.filePath.toString()))
+
+ try {
+ // Open a stream-based reader instead of materializing the entire file as byte[].
+ // This avoids the 2 GB byte[] limit and reduces memory pressure for large files.
+ currentImageStream = new HadoopImageInputStream(path, configuration)
+
+ // Create in-db GridCoverage2D from GeoTiff stream. The RenderedImage is lazy -
+ // pixel data will only be decoded when accessed via image.getData(Rectangle).
+ currentRaster = RasterConstructors.fromGeoTiff(currentImageStream)
+ currentIterator = rasterToInternalRows(currentRaster, dataSchema, rasterOptions, path)
+ currentFileIndex += 1
+ } catch {
+ case e: Exception =>
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+ throw e
+ }
+ }
+}
+
+object RasterPartitionReader {
+ def rasterToInternalRows(
+ currentRaster: GridCoverage2D,
+ dataSchema: StructType,
+ rasterOptions: RasterOptions,
+ path: Path): Iterator[InternalRow] = {
+ val retile = rasterOptions.retile
+ val tileWidth = rasterOptions.tileWidth
+ val tileHeight = rasterOptions.tileHeight
+ val padWithNoData = rasterOptions.padWithNoData
+
+ val writer = new UnsafeRowWriter(dataSchema.length)
+ writer.resetRowWriter()
+
+ // Extract the file name from the path
+ val fileName = path.getName
+
+ if (retile) {
+ val (tw, th) = (tileWidth, tileHeight) match {
+ case (Some(tw), Some(th)) => (tw, th)
+ case (None, None) =>
+ // Use the internal tile size of the input raster
+ val tw = currentRaster.getRenderedImage.getTileWidth
+ val th = currentRaster.getRenderedImage.getTileHeight
+ val tileSizeError = {
+ """To resolve this issue, you can try one of the following methods:
+ | 1. Disable retile by setting `.option("retile", "false")`.
+ | 2. Explicitly set `tileWidth` and `tileHeight`.
+ | 3. Convert the raster to a Cloud Optimized GeoTIFF (COG) using tools like `gdal_translate`.
+ |""".stripMargin
+ }
+ if (tw >= MAX_AUTO_TILE_SIZE || th >= MAX_AUTO_TILE_SIZE) {
+ throw new IllegalArgumentException(
+ s"Internal tile size of $path is too large ($tw x $th). " + tileSizeError)
+ }
+ if (tw == 0 || th == 0) {
+ throw new IllegalArgumentException(
+ s"Internal tile size of $path contains zero ($tw x $th). " + tileSizeError)
+ }
+ if (tw / th > 10 || th / tw > 10) {
Review Comment:
The “too thin” check uses integer division (`tw / th`), which truncates and can miss thin tiles (e.g., 4095/400 becomes 10 instead of 10.23). Use floating-point division (`tw.toDouble / th` and vice versa) to correctly enforce the aspect ratio constraint.
```suggestion
if (tw.toDouble / th > 10.0 || th.toDouble / tw > 10.0) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2851416976
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterDataSource.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.catalog.TableProvider
+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.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import scala.collection.JavaConverters._
+
+/**
+ * A Spark SQL data source for reading and writing raster images. This data source supports
+ * reading various raster formats as in-db rasters. Write support is implemented by falling back
+ * to the V1 data source [[RasterFileFormat]].
+ */
+class RasterDataSource extends FileDataSourceV2 with TableProvider with DataSourceRegister {
+
+ override def shortName(): String = "raster"
+
+ private def createRasterTable(
+ options: CaseInsensitiveStringMap,
+ userSchema: Option[StructType] = None): Table = {
+ var paths = getPaths(options)
+ var optionsWithoutPaths = getOptionsWithoutPaths(options)
+ val tableName = getTableName(options, paths)
+ val rasterOptions = new RasterOptions(optionsWithoutPaths.asScala.toMap)
+
+ if (paths.size == 1) {
+ if (paths.head.endsWith("/")) {
+ // Paths ends with / will be recursively loaded
+ val newOptions =
+ new java.util.HashMap[String, String](optionsWithoutPaths.asCaseSensitiveMap())
+ newOptions.put("recursiveFileLookup", "true")
+ if (!newOptions.containsKey("pathGlobFilter")) {
+ newOptions.put("pathGlobFilter", "*.{tif,tiff,TIF,TIFF}")
+ }
+ optionsWithoutPaths = new CaseInsensitiveStringMap(newOptions)
+ } else {
+ // Rewrite paths such as /path/to/some*glob*.tif into /path/to with
+ // pathGlobFilter="some*glob*.tif". This is for avoiding listing .tif
+ // files as directories when discovering files to load. Globs ends with
+ // .tif or .tiff should be files in the context of raster data loading.
+ val loadTifPattern = "(.*)/([^/]*\\*[^/]*\\.(?:tif|tiff))$".r
Review Comment:
The regex pattern only matches lowercase `.tif` and `.tiff` extensions, but line 55 shows that uppercase `.TIF` and `.TIFF` should also be supported (as indicated by the default pathGlobFilter). Paths like `/path/to/*.TIF` or `/path/to/*.TIFF` won't be rewritten correctly. Consider making the pattern case-insensitive by changing the regex to `"(.*)/([^/]*\\*[^/]*\\.(?i:tif|tiff))$".r` or `"(.*)/([^/]*\\*[^/]*\\.[tT][iI][fF]{1,2})$".r`.
```suggestion
val loadTifPattern = "(.*)/([^/]*\\*[^/]*\\.(?i:tif|tiff))$".r
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "jiayuasu (via GitHub)" <gi...@apache.org>.
jiayuasu commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2851457817
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterDataSource.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.catalog.TableProvider
+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.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import scala.collection.JavaConverters._
+
+/**
+ * A Spark SQL data source for reading and writing raster images. This data source supports
+ * reading various raster formats as in-db rasters. Write support is implemented by falling back
+ * to the V1 data source [[RasterFileFormat]].
+ */
+class RasterDataSource extends FileDataSourceV2 with TableProvider with DataSourceRegister {
+
+ override def shortName(): String = "raster"
+
+ private def createRasterTable(
+ options: CaseInsensitiveStringMap,
+ userSchema: Option[StructType] = None): Table = {
+ var paths = getPaths(options)
+ var optionsWithoutPaths = getOptionsWithoutPaths(options)
+ val tableName = getTableName(options, paths)
+ val rasterOptions = new RasterOptions(optionsWithoutPaths.asScala.toMap)
+
+ if (paths.size == 1) {
+ if (paths.head.endsWith("/")) {
+ // Paths ends with / will be recursively loaded
+ val newOptions =
+ new java.util.HashMap[String, String](optionsWithoutPaths.asCaseSensitiveMap())
+ newOptions.put("recursiveFileLookup", "true")
+ if (!newOptions.containsKey("pathGlobFilter")) {
+ newOptions.put("pathGlobFilter", "*.{tif,tiff,TIF,TIFF}")
+ }
+ optionsWithoutPaths = new CaseInsensitiveStringMap(newOptions)
+ } else {
+ // Rewrite paths such as /path/to/some*glob*.tif into /path/to with
+ // pathGlobFilter="some*glob*.tif". This is for avoiding listing .tif
+ // files as directories when discovering files to load. Globs ends with
+ // .tif or .tiff should be files in the context of raster data loading.
+ val loadTifPattern = "(.*)/([^/]*\\*[^/]*\\.(?:tif|tiff))$".r
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2845919143
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterPartitionReader.scala:
##########
@@ -0,0 +1,213 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.sedona.common.raster.RasterConstructors
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter
+import org.apache.spark.sql.connector.read.PartitionReader
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.sedona_sql.UDT.RasterUDT
+import org.apache.spark.sql.sedona_sql.io.raster.RasterPartitionReader.rasterToInternalRows
+import org.apache.spark.sql.sedona_sql.io.raster.RasterTable.{MAX_AUTO_TILE_SIZE, RASTER, RASTER_NAME, TILE_X, TILE_Y}
+import org.apache.spark.sql.types.StructType
+import org.geotools.coverage.grid.GridCoverage2D
+
+import java.net.URI
+import scala.collection.JavaConverters._
+
+class RasterPartitionReader(
+ configuration: Configuration,
+ partitionedFiles: Array[PartitionedFile],
+ dataSchema: StructType,
+ rasterOptions: RasterOptions)
+ extends PartitionReader[InternalRow] {
+
+ // Track the current file index we're processing
+ private var currentFileIndex = 0
+
+ // Current raster being processed
+ private var currentRaster: GridCoverage2D = _
+
+ // Current row
+ private var currentRow: InternalRow = _
+
+ // Iterator for the current file's tiles
+ private var currentIterator: Iterator[InternalRow] = Iterator.empty
+
+ override def next(): Boolean = {
+ // If current iterator has more elements, return true
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+
+ // If current iterator is exhausted, but we have more files, load the next file
+ if (currentFileIndex < partitionedFiles.length) {
+ loadNextFile()
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+ }
+
+ // No more data
+ false
+ }
+
+ override def get(): InternalRow = {
+ currentRow
+ }
+
+ override def close(): Unit = {
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ }
+
+ private def loadNextFile(): Unit = {
+ // Clean up previous raster if exists
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+
+ if (currentFileIndex >= partitionedFiles.length) {
+ currentIterator = Iterator.empty
+ return
+ }
+
+ val partition = partitionedFiles(currentFileIndex)
+ val path = new Path(new URI(partition.filePath.toString()))
+
+ try {
+ // Read file bytes from Hadoop FS
+ val fs = path.getFileSystem(configuration)
+ val fileStatus = fs.getFileStatus(path)
+ val fileLength = fileStatus.getLen.toInt
+ val bytes = new Array[Byte](fileLength)
+ val inputStream = fs.open(path)
+ try {
+ org.apache.hadoop.io.IOUtils.readFully(inputStream, bytes, 0, fileLength)
+ } finally {
+ inputStream.close()
+ }
+
+ // Create in-db GridCoverage2D from GeoTiff bytes. The RenderedImage is lazy -
+ // pixel data will only be decoded when accessed via image.getData(Rectangle).
+ currentRaster = RasterConstructors.fromGeoTiff(bytes)
+ currentIterator = rasterToInternalRows(currentRaster, dataSchema, rasterOptions, path)
Review Comment:
This reader materializes the entire file into a single `Array[Byte]` and casts `getLen` to `Int`. For GeoTIFFs >2GB this will overflow / fail (and still hits Java array size limits), which undermines the stated goal of bypassing Spark’s 2GB record limit. Consider streaming the GeoTIFF via an `InputStream`/`ImageInputStream`-backed reader (or at least detect `getLen > Int.MaxValue` and fail with a clear error).
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterScanBuilder.scala:
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.read.Batch
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.connector.read.PartitionReaderFactory
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.connector.read.SupportsPushDownLimit
+import org.apache.spark.sql.connector.read.SupportsPushDownTableSample
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.v2.FileScanBuilder
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.execution.datasources.v2.FileScan
+import org.apache.spark.sql.execution.datasources.v2.TableSampleInfo
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.util.SerializableConfiguration
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+import scala.util.Random
+
+case class RasterScanBuilder(
+ sparkSession: SparkSession,
+ fileIndex: PartitioningAwareFileIndex,
+ schema: StructType,
+ dataSchema: StructType,
+ options: CaseInsensitiveStringMap,
+ rasterOptions: RasterOptions)
+ extends FileScanBuilder(sparkSession, fileIndex, dataSchema)
+ with SupportsPushDownTableSample
+ with SupportsPushDownLimit {
+
+ private var pushedTableSample: Option[TableSampleInfo] = None
+ private var pushedLimit: Option[Int] = None
+
+ override def build(): Scan = {
+ RasterScan(
+ sparkSession,
+ fileIndex,
+ dataSchema,
+ readDataSchema(),
+ readPartitionSchema(),
+ options,
+ rasterOptions,
+ pushedDataFilters,
+ partitionFilters,
+ dataFilters,
+ pushedTableSample,
+ pushedLimit)
+ }
+
+ override def pushTableSample(
+ lowerBound: Double,
+ upperBound: Double,
+ withReplacement: Boolean,
+ seed: Long): Boolean = {
+ if (withReplacement || rasterOptions.retile) {
+ false
+ } else {
+ pushedTableSample = Some(TableSampleInfo(lowerBound, upperBound, withReplacement, seed))
+ true
+ }
+ }
+
+ override def pushLimit(limit: Int): Boolean = {
+ pushedLimit = Some(limit)
+ true
+ }
+
+ override def isPartiallyPushed: Boolean = rasterOptions.retile
+}
+
+case class RasterScan(
+ sparkSession: SparkSession,
+ fileIndex: PartitioningAwareFileIndex,
+ dataSchema: StructType,
+ readDataSchema: StructType,
+ readPartitionSchema: StructType,
+ options: CaseInsensitiveStringMap,
+ rasterOptions: RasterOptions,
+ pushedFilters: Array[Filter],
+ partitionFilters: Seq[Expression] = Seq.empty,
+ dataFilters: Seq[Expression] = Seq.empty,
+ pushedTableSample: Option[TableSampleInfo] = None,
+ pushedLimit: Option[Int] = None)
+ extends FileScan
+ with Batch {
+
+ private lazy val inputPartitions = {
+ var partitions = super.planInputPartitions()
+
+ // Sample the files based on the table sample
+ pushedTableSample.foreach { tableSample =>
+ val r = new Random(tableSample.seed)
+ var partitionIndex = 0
+ partitions = partitions.flatMap {
+ case filePartition: FilePartition =>
+ val files = filePartition.files
+ val sampledFiles = files.filter(_ => r.nextDouble() < tableSample.upperBound)
+ if (sampledFiles.nonEmpty) {
+ val index = partitionIndex
+ partitionIndex += 1
+ Some(FilePartition(index, sampledFiles))
+ } else {
+ None
+ }
+ case partition =>
+ throw new IllegalArgumentException(
+ s"Unexpected partition type: ${partition.getClass.getCanonicalName}")
Review Comment:
`pushTableSample` receives both `lowerBound` and `upperBound`, but the sampling logic only uses `upperBound`. This changes Spark’s table-sample semantics for non-zero lower bounds. The filter should accept a file iff the random value is in `[lowerBound, upperBound)` (and still respect `withReplacement`).
```suggestion
// Only push down sampling for without-replacement semantics.
if (!tableSample.withReplacement) {
val r = new Random(tableSample.seed)
var partitionIndex = 0
partitions = partitions.flatMap {
case filePartition: FilePartition =>
val files = filePartition.files
val sampledFiles = files.filter { _ =>
val v = r.nextDouble()
v >= tableSample.lowerBound && v < tableSample.upperBound
}
if (sampledFiles.nonEmpty) {
val index = partitionIndex
partitionIndex += 1
Some(FilePartition(index, sampledFiles))
} else {
None
}
case partition =>
throw new IllegalArgumentException(
s"Unexpected partition type: ${partition.getClass.getCanonicalName}")
}
```
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterTable.scala:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.hadoop.fs.FileStatus
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.connector.catalog.SupportsRead
+import org.apache.spark.sql.connector.catalog.SupportsWrite
+import org.apache.spark.sql.connector.catalog.TableCapability
+import org.apache.spark.sql.connector.read.ScanBuilder
+import org.apache.spark.sql.connector.write.LogicalWriteInfo
+import org.apache.spark.sql.connector.write.WriteBuilder
+import org.apache.spark.sql.execution.datasources.v2.FileTable
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.sedona_sql.UDT.RasterUDT
+import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.sql.types.StructField
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import java.util.{Set => JSet}
+
+case class RasterTable(
+ name: String,
+ sparkSession: SparkSession,
+ options: CaseInsensitiveStringMap,
+ paths: Seq[String],
+ userSpecifiedSchema: Option[StructType],
+ rasterOptions: RasterOptions,
+ fallbackFileFormat: Class[_ <: FileFormat])
+ extends FileTable(sparkSession, options, paths, userSpecifiedSchema)
+ with SupportsRead
+ with SupportsWrite {
+
+ override def inferSchema(files: Seq[FileStatus]): Option[StructType] =
+ Some(userSpecifiedSchema.getOrElse(RasterTable.inferSchema(rasterOptions)))
+
+ override def formatName: String = "Raster"
+
+ override def capabilities(): JSet[TableCapability] =
+ java.util.EnumSet.of(TableCapability.BATCH_READ)
+
+ override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = {
+ RasterScanBuilder(sparkSession, fileIndex, schema, dataSchema, options, rasterOptions)
+ }
+
+ override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder =
+ // Note: this code path will never be taken, since Spark will always fall back to V1
+ // data source when writing to File source v2. See SPARK-28396: File source v2 write
+ // path is currently broken.
+ null
Review Comment:
`newWriteBuilder` returns `null` while the table mixes in `SupportsWrite`. If Spark ever calls this code path, this will cause an NPE. Prefer either (a) removing `SupportsWrite` entirely since `capabilities()` only advertises `BATCH_READ`, or (b) returning a valid `WriteBuilder` / throwing a clear `UnsupportedOperationException`.
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/raster/RasterConstructors.scala:
##########
@@ -107,39 +109,35 @@ private[apache] case class RS_TileExplode(children: Seq[Expression])
override def eval(input: InternalRow): TraversableOnce[InternalRow] = {
val raster = arguments.rasterExpr.toRaster(input)
- try {
- val bandIndices = arguments.bandIndicesExpr.eval(input).asInstanceOf[ArrayData] match {
- case null => null
- case value: Any => value.toIntArray
- }
- val tileWidth = arguments.tileWidthExpr.eval(input).asInstanceOf[Int]
- val tileHeight = arguments.tileHeightExpr.eval(input).asInstanceOf[Int]
- val padWithNoDataValue = arguments.padWithNoDataExpr.eval(input).asInstanceOf[Boolean]
- val noDataValue = arguments.noDataValExpr.eval(input) match {
- case null => Double.NaN
- case value: Integer => value.toDouble
- case value: Decimal => value.toDouble
- case value: Float => value.toDouble
- case value: Double => value
- case value: Any =>
- throw new IllegalArgumentException(
- "Unsupported class for noDataValue: " + value.getClass)
- }
- val tiles = RasterConstructors.generateTiles(
- raster,
- bandIndices,
- tileWidth,
- tileHeight,
- padWithNoDataValue,
- noDataValue)
- tiles.map { tile =>
- val gridCoverage2D = tile.getCoverage
- val row = InternalRow(tile.getTileX, tile.getTileY, gridCoverage2D.serialize)
- gridCoverage2D.dispose(true)
- row
- }
- } finally {
- raster.dispose(true)
+ val bandIndices = arguments.bandIndicesExpr.eval(input).asInstanceOf[ArrayData] match {
+ case null => null
+ case value: Any => value.toIntArray
+ }
+ val tileWidth = arguments.tileWidthExpr.eval(input).asInstanceOf[Int]
+ val tileHeight = arguments.tileHeightExpr.eval(input).asInstanceOf[Int]
+ val padWithNoDataValue = arguments.padWithNoDataExpr.eval(input).asInstanceOf[Boolean]
+ val noDataValue = arguments.noDataValExpr.eval(input) match {
+ case null => Double.NaN
+ case value: Integer => value.toDouble
+ case value: Decimal => value.toDouble
+ case value: Float => value.toDouble
+ case value: Double => value
+ case value: Any =>
+ throw new IllegalArgumentException("Unsupported class for noDataValue: " + value.getClass)
+ }
+ val tileIterator = RasterConstructors.generateTiles(
+ raster,
+ bandIndices,
+ tileWidth,
+ tileHeight,
+ padWithNoDataValue,
+ noDataValue)
+ tileIterator.setAutoDisposeSource(true)
+ tileIterator.asScala.map { tile =>
+ val gridCoverage2D = tile.getCoverage
+ val row = InternalRow(tile.getTileX, tile.getTileY, gridCoverage2D.serialize)
+ gridCoverage2D.dispose(true)
+ row
Review Comment:
`RS_TileExplode` no longer disposes the source raster in a `finally`. With the new lazy iterator, the source is only disposed when the iterator reaches the end; if iteration is short-circuited (e.g., by `limit`/cancellation) the raster can leak. Consider wiring a disposal hook that runs when Spark stops consuming the iterator (e.g., using a `CompletionIterator`/task completion listener), or otherwise ensure disposal even on early termination.
##########
common/src/main/java/org/apache/sedona/common/raster/TileGenerator.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.common.raster;
+
+import java.awt.Rectangle;
+import java.awt.image.Raster;
+import java.awt.image.RenderedImage;
+import java.awt.image.WritableRaster;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import javax.media.jai.RasterFactory;
+import org.apache.sedona.common.utils.ImageUtils;
+import org.apache.sedona.common.utils.RasterUtils;
+import org.geotools.api.metadata.spatial.PixelOrientation;
+import org.geotools.api.referencing.datum.PixelInCell;
+import org.geotools.coverage.GridSampleDimension;
+import org.geotools.coverage.grid.GridCoverage2D;
+import org.geotools.coverage.grid.GridEnvelope2D;
+import org.geotools.coverage.grid.GridGeometry2D;
+import org.geotools.referencing.operation.transform.AffineTransform2D;
+
+public class TileGenerator {
+
+ public static class Tile {
+
+ private final int tileX;
+ private final int tileY;
+ private final GridCoverage2D coverage;
+
+ public Tile(int tileX, int tileY, GridCoverage2D coverage) {
+ this.tileX = tileX;
+ this.tileY = tileY;
+ this.coverage = coverage;
+ }
+
+ public int getTileX() {
+ return tileX;
+ }
+
+ public int getTileY() {
+ return tileY;
+ }
+
+ public GridCoverage2D getCoverage() {
+ return coverage;
+ }
+ }
+
+ /**
+ * Generate tiles from an in-db grid coverage. The generated tiles are also in-db grid coverages.
+ * Pixel data will be copied into the tiles one tile at a time via a lazy iterator.
+ *
+ * @param gridCoverage2D the in-db grid coverage
+ * @param bandIndices the indices of the bands to select (1-based)
+ * @param tileWidth the width of the tiles
+ * @param tileHeight the height of the tiles
+ * @param padWithNoData whether to pad the tiles with no data value
+ * @param padNoDataValue the no data value for padded tiles, only used when padWithNoData is true.
+ * If the value is NaN, the no data value of the original band will be used.
+ * @return a lazy iterator of tiles
+ */
+ public static InDbTileIterator generateInDbTiles(
+ GridCoverage2D gridCoverage2D,
+ int[] bandIndices,
+ int tileWidth,
+ int tileHeight,
+ boolean padWithNoData,
+ double padNoDataValue) {
+ return new InDbTileIterator(
+ gridCoverage2D, bandIndices, tileWidth, tileHeight, padWithNoData, padNoDataValue);
+ }
+
+ public abstract static class TileIterator implements Iterator<Tile> {
+ protected final GridCoverage2D gridCoverage2D;
+ protected int numTileX;
+ protected int numTileY;
+ protected int tileX;
+ protected int tileY;
+ protected boolean autoDisposeSource = false;
+ protected Runnable disposeFunction = null;
+
+ TileIterator(GridCoverage2D gridCoverage2D) {
+ this.gridCoverage2D = gridCoverage2D;
+ }
+
+ protected void initialize(int numTileX, int numTileY) {
+ this.numTileX = numTileX;
+ this.numTileY = numTileY;
+ this.tileX = 0;
+ this.tileY = 0;
+ }
+
+ /**
+ * Set whether to dispose the grid coverage when the iterator reaches the end. Default is false.
+ *
+ * @param autoDisposeSource whether to dispose the grid coverage
+ */
+ public void setAutoDisposeSource(boolean autoDisposeSource) {
+ this.autoDisposeSource = autoDisposeSource;
+ }
+
+ public void setDisposeFunction(Runnable disposeFunction) {
+ this.disposeFunction = disposeFunction;
+ }
+
+ public int getNumTileX() {
+ return numTileX;
+ }
+
+ public int getNumTileY() {
+ return numTileY;
+ }
+
+ public int getNumTiles() {
+ return numTileX * numTileY;
+ }
+
+ protected abstract Tile generateTile();
+
+ @Override
+ public boolean hasNext() {
+ if (numTileX == 0 || numTileY == 0) {
+ return false;
+ }
+ return tileY < numTileY;
+ }
+
+ @Override
+ public Tile next() {
+ // Check if current tile coordinate is valid
+ if (tileX >= numTileX || tileY >= numTileY) {
+ throw new NoSuchElementException();
+ }
+
+ Tile tile = generateTile();
+
+ // Advance to the next tile
+ tileX += 1;
+ if (tileX >= numTileX) {
+ tileX = 0;
+ tileY += 1;
+
+ // Dispose the grid coverage if we are at the end
+ if (tileY >= numTileY) {
+ if (autoDisposeSource) {
+ gridCoverage2D.dispose(true);
+ }
+ if (disposeFunction != null) {
+ disposeFunction.run();
+ }
+ }
+ }
+
+ return tile;
+ }
+ }
+
+ public static class InDbTileIterator extends TileIterator {
+ private final int[] bandIndices;
+ private final int tileWidth;
+ private final int tileHeight;
+ private final boolean padWithNoData;
+ private final double padNoDataValue;
+ private final AffineTransform2D affine;
+ private final RenderedImage image;
+ private final double[] noDataValues;
+ private final int imageWidth;
+ private final int imageHeight;
+
+ public InDbTileIterator(
+ GridCoverage2D gridCoverage2D,
+ int[] bandIndices,
+ int tileWidth,
+ int tileHeight,
+ boolean padWithNoData,
+ double padNoDataValue) {
+ super(gridCoverage2D);
+ this.bandIndices = bandIndices;
+ this.tileWidth = tileWidth;
+ this.tileHeight = tileHeight;
+ this.padWithNoData = padWithNoData;
+ this.padNoDataValue = padNoDataValue;
+
+ affine = RasterUtils.getAffineTransform(gridCoverage2D, PixelOrientation.CENTER);
+ image = gridCoverage2D.getRenderedImage();
+ noDataValues = new double[bandIndices.length];
+ for (int i = 0; i < bandIndices.length; i++) {
+ noDataValues[i] =
+ RasterUtils.getNoDataValue(gridCoverage2D.getSampleDimension(bandIndices[i] - 1));
+ }
+ imageWidth = image.getWidth();
+ imageHeight = image.getHeight();
+ int numTileX = (int) Math.ceil((double) imageWidth / tileWidth);
+ int numTileY = (int) Math.ceil((double) imageHeight / tileHeight);
+ initialize(numTileX, numTileY);
+ }
+
+ @Override
+ protected Tile generateTile() {
+ // Process the current tile
+ int x0 = tileX * tileWidth + image.getMinX();
+ int y0 = tileY * tileHeight + image.getMinY();
+
+ // Rect to copy from the original image
+ int rectWidth = Math.min(tileWidth, imageWidth - x0);
+ int rectHeight = Math.min(tileHeight, imageHeight - y0);
Review Comment:
`x0`/`y0` are offset by `image.getMinX/getMinY`, but `rectWidth`/`rectHeight` are computed using `imageWidth - x0` / `imageHeight - y0`, which is incorrect when `minX/minY` are non-zero and can truncate tiles. Compute remaining width/height using `(image.getMinX + imageWidth) - x0` (and similarly for Y), or compute `x0/y0` in the same coordinate system as `imageWidth/imageHeight`.
```suggestion
int rectWidth =
Math.min(tileWidth, (image.getMinX() + imageWidth) - x0);
int rectHeight =
Math.min(tileHeight, (image.getMinY() + imageHeight) - y0);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "jiayuasu (via GitHub)" <gi...@apache.org>.
jiayuasu commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2851276883
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterPartitionReader.scala:
##########
@@ -0,0 +1,213 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.sedona.common.raster.RasterConstructors
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter
+import org.apache.spark.sql.connector.read.PartitionReader
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.sedona_sql.UDT.RasterUDT
+import org.apache.spark.sql.sedona_sql.io.raster.RasterPartitionReader.rasterToInternalRows
+import org.apache.spark.sql.sedona_sql.io.raster.RasterTable.{MAX_AUTO_TILE_SIZE, RASTER, RASTER_NAME, TILE_X, TILE_Y}
+import org.apache.spark.sql.types.StructType
+import org.geotools.coverage.grid.GridCoverage2D
+
+import java.net.URI
+import scala.collection.JavaConverters._
+
+class RasterPartitionReader(
+ configuration: Configuration,
+ partitionedFiles: Array[PartitionedFile],
+ dataSchema: StructType,
+ rasterOptions: RasterOptions)
+ extends PartitionReader[InternalRow] {
+
+ // Track the current file index we're processing
+ private var currentFileIndex = 0
+
+ // Current raster being processed
+ private var currentRaster: GridCoverage2D = _
+
+ // Current row
+ private var currentRow: InternalRow = _
+
+ // Iterator for the current file's tiles
+ private var currentIterator: Iterator[InternalRow] = Iterator.empty
+
+ override def next(): Boolean = {
+ // If current iterator has more elements, return true
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+
+ // If current iterator is exhausted, but we have more files, load the next file
+ if (currentFileIndex < partitionedFiles.length) {
+ loadNextFile()
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+ }
+
+ // No more data
+ false
+ }
+
+ override def get(): InternalRow = {
+ currentRow
+ }
+
+ override def close(): Unit = {
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ }
+
+ private def loadNextFile(): Unit = {
+ // Clean up previous raster if exists
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+
+ if (currentFileIndex >= partitionedFiles.length) {
+ currentIterator = Iterator.empty
+ return
+ }
+
+ val partition = partitionedFiles(currentFileIndex)
+ val path = new Path(new URI(partition.filePath.toString()))
+
+ try {
+ // Read file bytes from Hadoop FS
+ val fs = path.getFileSystem(configuration)
+ val fileStatus = fs.getFileStatus(path)
+ val fileLength = fileStatus.getLen.toInt
+ val bytes = new Array[Byte](fileLength)
+ val inputStream = fs.open(path)
+ try {
+ org.apache.hadoop.io.IOUtils.readFully(inputStream, bytes, 0, fileLength)
+ } finally {
+ inputStream.close()
+ }
+
+ // Create in-db GridCoverage2D from GeoTiff bytes. The RenderedImage is lazy -
+ // pixel data will only be decoded when accessed via image.getData(Rectangle).
+ currentRaster = RasterConstructors.fromGeoTiff(bytes)
+ currentIterator = rasterToInternalRows(currentRaster, dataSchema, rasterOptions, path)
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "jiayuasu (via GitHub)" <gi...@apache.org>.
jiayuasu commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2851276218
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterScanBuilder.scala:
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.read.Batch
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.connector.read.PartitionReaderFactory
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.connector.read.SupportsPushDownLimit
+import org.apache.spark.sql.connector.read.SupportsPushDownTableSample
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.v2.FileScanBuilder
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.execution.datasources.v2.FileScan
+import org.apache.spark.sql.execution.datasources.v2.TableSampleInfo
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.util.SerializableConfiguration
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+import scala.util.Random
+
+case class RasterScanBuilder(
+ sparkSession: SparkSession,
+ fileIndex: PartitioningAwareFileIndex,
+ schema: StructType,
+ dataSchema: StructType,
+ options: CaseInsensitiveStringMap,
+ rasterOptions: RasterOptions)
+ extends FileScanBuilder(sparkSession, fileIndex, dataSchema)
+ with SupportsPushDownTableSample
+ with SupportsPushDownLimit {
+
+ private var pushedTableSample: Option[TableSampleInfo] = None
+ private var pushedLimit: Option[Int] = None
+
+ override def build(): Scan = {
+ RasterScan(
+ sparkSession,
+ fileIndex,
+ dataSchema,
+ readDataSchema(),
+ readPartitionSchema(),
+ options,
+ rasterOptions,
+ pushedDataFilters,
+ partitionFilters,
+ dataFilters,
+ pushedTableSample,
+ pushedLimit)
+ }
+
+ override def pushTableSample(
+ lowerBound: Double,
+ upperBound: Double,
+ withReplacement: Boolean,
+ seed: Long): Boolean = {
+ if (withReplacement || rasterOptions.retile) {
+ false
+ } else {
+ pushedTableSample = Some(TableSampleInfo(lowerBound, upperBound, withReplacement, seed))
+ true
+ }
+ }
+
+ override def pushLimit(limit: Int): Boolean = {
+ pushedLimit = Some(limit)
+ true
+ }
+
+ override def isPartiallyPushed: Boolean = rasterOptions.retile
+}
+
+case class RasterScan(
+ sparkSession: SparkSession,
+ fileIndex: PartitioningAwareFileIndex,
+ dataSchema: StructType,
+ readDataSchema: StructType,
+ readPartitionSchema: StructType,
+ options: CaseInsensitiveStringMap,
+ rasterOptions: RasterOptions,
+ pushedFilters: Array[Filter],
+ partitionFilters: Seq[Expression] = Seq.empty,
+ dataFilters: Seq[Expression] = Seq.empty,
+ pushedTableSample: Option[TableSampleInfo] = None,
+ pushedLimit: Option[Int] = None)
+ extends FileScan
+ with Batch {
+
+ private lazy val inputPartitions = {
+ var partitions = super.planInputPartitions()
+
+ // Sample the files based on the table sample
+ pushedTableSample.foreach { tableSample =>
+ val r = new Random(tableSample.seed)
+ var partitionIndex = 0
+ partitions = partitions.flatMap {
+ case filePartition: FilePartition =>
+ val files = filePartition.files
+ val sampledFiles = files.filter(_ => r.nextDouble() < tableSample.upperBound)
+ if (sampledFiles.nonEmpty) {
+ val index = partitionIndex
+ partitionIndex += 1
+ Some(FilePartition(index, sampledFiles))
+ } else {
+ None
+ }
+ case partition =>
+ throw new IllegalArgumentException(
+ s"Unexpected partition type: ${partition.getClass.getCanonicalName}")
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "jiayuasu (via GitHub)" <gi...@apache.org>.
jiayuasu commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2851395005
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterOptions.scala:
##########
@@ -20,16 +20,52 @@ package org.apache.spark.sql.sedona_sql.io.raster
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
-private[io] class RasterOptions(@transient private val parameters: CaseInsensitiveMap[String])
+class RasterOptions(@transient private val parameters: CaseInsensitiveMap[String])
extends Serializable {
def this(parameters: Map[String, String]) = this(CaseInsensitiveMap(parameters))
+ // The following options are used to read raster data
+
+ /**
+ * Whether to retile the raster data. If true, the raster data will be retiled into smaller
+ * tiles. If false, the raster data will be read as a single tile.
+ */
+ val retile: Boolean = parameters.getOrElse("retile", "true").toBoolean
+
+ /**
+ * The width of the tile. This is only effective when retile is true. If retile is true and
+ * tileWidth is not set, the default value is the width of the internal tiles in the raster
+ * files. Each raster file may have different internal tile sizes.
+ */
+ val tileWidth: Option[Int] = parameters.get("tileWidth").map(_.toInt)
+
+ /**
+ * The height of the tile. This is only effective when retile is true. If retile is true and
+ * tileHeight is not set, the default value is the same as tileWidth. If tileHeight is set,
+ * tileWidth must be set as well.
+ */
+ val tileHeight: Option[Int] = parameters
+ .get("tileHeight")
+ .map { value =>
+ require(tileWidth.isDefined, "tileWidth must be set when tileHeight is set")
+ value.toInt
+ }
+ .orElse(tileWidth)
+
Review Comment:
Fixed
##########
common/src/main/java/org/apache/sedona/common/raster/inputstream/HadoopImageInputStream.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.common.raster.inputstream;
+
+import java.io.IOException;
+import javax.imageio.stream.ImageInputStreamImpl;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+/** An ImageInputStream that reads image data from a Hadoop FileSystem. */
+public class HadoopImageInputStream extends ImageInputStreamImpl {
+
+ private final FSDataInputStream stream;
+ private final Path path;
+ private final Configuration conf;
+
+ public HadoopImageInputStream(Path path, Configuration conf) throws IOException {
+ FileSystem fs = path.getFileSystem(conf);
+ stream = fs.open(path);
+ this.path = path;
+ this.conf = conf;
+ }
+
+ public HadoopImageInputStream(Path path) throws IOException {
+ this(path, new Configuration());
+ }
+
+ public HadoopImageInputStream(FSDataInputStream stream) {
+ this.stream = stream;
+ this.path = null;
+ this.conf = null;
+ }
+
+ public Path getPath() {
+ return path;
+ }
+
+ public Configuration getConf() {
+ return conf;
+ }
+
+ @Override
+ public void close() throws IOException {
+ super.close();
+ stream.close();
+ }
+
+ @Override
+ public int read() throws IOException {
+ byte[] buf = new byte[1];
+ int ret_len = read(buf, 0, 1);
+ if (ret_len < 0) {
+ return ret_len;
+ }
+ return buf[0] & 0xFF;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ checkClosed();
+ bitOffset = 0;
+
+ if (len == 0) {
+ return 0;
+ }
+
+ // stream.read may return fewer data than requested, so we need to loop until we get all the
+ // data, or hit the end of the stream. We can not simply perform an incomplete read and return
+ // the number of bytes actually read, since the methods in ImageInputStreamImpl such as
+ // readInt() relies on this method and assumes that partial reads only happens when reading
+ // EOF. This might be a bug of imageio since they should invoke readFully() in such cases.
+ int remaining = len;
+ while (remaining > 0) {
+ int ret_len = stream.read(b, off, remaining);
+ if (ret_len < 0) {
+ // Hit EOF, no more data to read.
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GH-2672] Add a new raster data source reader that can automatically tile GeoTiffs and bypass the Spark record limit [sedona]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2673:
URL: https://github.com/apache/sedona/pull/2673#discussion_r2851729930
##########
spark/common/src/test/scala/org/apache/sedona/sql/rasterIOTest.scala:
##########
@@ -239,5 +248,282 @@ class rasterIOTest extends TestBaseScala with BeforeAndAfter with GivenWhenThen
}
}
- override def afterAll(): Unit = FileUtils.deleteDirectory(new File(tempDir))
+ describe("Raster read test") {
+ it("should read geotiff using raster source with explicit tiling") {
+ val rasterDf = sparkSession.read
+ .format("raster")
+ .options(Map("retile" -> "true", "tileWidth" -> "64"))
+ .load(rasterdatalocation)
+ assert(rasterDf.count() > 100)
+ rasterDf.collect().foreach { row =>
Review Comment:
This test triggers two full scans of the raster source (`count()` then `collect()`), which increases CI time and doubles IO for no added coverage. Consider collecting once (or using `take`) and asserting on the collected row count instead of calling `count()` separately.
```suggestion
val rasterRows = rasterDf.collect()
assert(rasterRows.length > 100)
rasterRows.foreach { row =>
```
##########
spark/common/src/test/scala/org/apache/sedona/sql/rasterIOTest.scala:
##########
@@ -239,5 +248,282 @@ class rasterIOTest extends TestBaseScala with BeforeAndAfter with GivenWhenThen
}
}
- override def afterAll(): Unit = FileUtils.deleteDirectory(new File(tempDir))
+ describe("Raster read test") {
+ it("should read geotiff using raster source with explicit tiling") {
+ val rasterDf = sparkSession.read
+ .format("raster")
+ .options(Map("retile" -> "true", "tileWidth" -> "64"))
+ .load(rasterdatalocation)
+ assert(rasterDf.count() > 100)
+ rasterDf.collect().foreach { row =>
+ val raster = row.getAs[Object](0).asInstanceOf[GridCoverage2D]
+ assert(raster.getGridGeometry.getGridRange2D.width <= 64)
+ assert(raster.getGridGeometry.getGridRange2D.height <= 64)
+ val x = row.getInt(1)
+ val y = row.getInt(2)
+ assert(x >= 0 && y >= 0)
+ raster.dispose(true)
+ }
+
+ // Test projection push-down
+ rasterDf.selectExpr("y", "rast as r").collect().foreach { row =>
+ val raster = row.getAs[Object](1).asInstanceOf[GridCoverage2D]
+ assert(raster.getGridGeometry.getGridRange2D.width <= 64)
+ assert(raster.getGridGeometry.getGridRange2D.height <= 64)
+ val y = row.getInt(0)
+ assert(y >= 0)
+ raster.dispose(true)
+ }
+ }
+
+ it("should tile geotiff using raster source with padding enabled") {
+ val rasterDf = sparkSession.read
+ .format("raster")
+ .options(Map("retile" -> "true", "tileWidth" -> "64", "padWithNoData" -> "true"))
+ .load(rasterdatalocation)
+ assert(rasterDf.count() > 100)
+ rasterDf.collect().foreach { row =>
Review Comment:
This test also performs `count()` followed by `collect()`, causing two actions / scans. Consider a single action (e.g., `val rows = collect()` and assert on `rows.length`) to keep the test runtime and IO down.
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterPartitionReader.scala:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.sedona.common.raster.RasterConstructors
+import org.apache.sedona.common.raster.inputstream.HadoopImageInputStream
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter
+import org.apache.spark.sql.connector.read.PartitionReader
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.sedona_sql.UDT.RasterUDT
+import org.apache.spark.sql.sedona_sql.io.raster.RasterPartitionReader.rasterToInternalRows
+import org.apache.spark.sql.sedona_sql.io.raster.RasterTable.{MAX_AUTO_TILE_SIZE, RASTER, RASTER_NAME, TILE_X, TILE_Y}
+import org.apache.spark.sql.types.StructType
+import org.geotools.coverage.grid.GridCoverage2D
+
+import java.net.URI
+import scala.collection.JavaConverters._
+
+class RasterPartitionReader(
+ configuration: Configuration,
+ partitionedFiles: Array[PartitionedFile],
+ dataSchema: StructType,
+ rasterOptions: RasterOptions)
+ extends PartitionReader[InternalRow] {
+
+ // Track the current file index we're processing
+ private var currentFileIndex = 0
+
+ // Current raster being processed
+ private var currentRaster: GridCoverage2D = _
+
+ // Current image input stream (must be kept open while the raster is in use)
+ private var currentImageStream: HadoopImageInputStream = _
+
+ // Current row
+ private var currentRow: InternalRow = _
+
+ // Iterator for the current file's tiles
+ private var currentIterator: Iterator[InternalRow] = Iterator.empty
+
+ override def next(): Boolean = {
+ // If current iterator has more elements, return true
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+
+ // If current iterator is exhausted, but we have more files, load the next file
+ if (currentFileIndex < partitionedFiles.length) {
+ loadNextFile()
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+ }
+
+ // No more data
+ false
+ }
+
+ override def get(): InternalRow = {
+ currentRow
+ }
+
+ override def close(): Unit = {
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+ }
+
+ private def loadNextFile(): Unit = {
+ // Clean up previous raster and stream if exists
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+
+ if (currentFileIndex >= partitionedFiles.length) {
+ currentIterator = Iterator.empty
+ return
+ }
+
+ val partition = partitionedFiles(currentFileIndex)
+ val path = new Path(new URI(partition.filePath.toString()))
+
+ try {
+ // Open a stream-based reader instead of materializing the entire file as byte[].
+ // This avoids the 2 GB byte[] limit and reduces memory pressure for large files.
+ currentImageStream = new HadoopImageInputStream(path, configuration)
+
+ // Create in-db GridCoverage2D from GeoTiff stream. The RenderedImage is lazy -
+ // pixel data will only be decoded when accessed via image.getData(Rectangle).
+ currentRaster = RasterConstructors.fromGeoTiff(currentImageStream)
+ currentIterator = rasterToInternalRows(currentRaster, dataSchema, rasterOptions, path)
+ currentFileIndex += 1
+ } catch {
+ case e: Exception =>
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+ throw e
+ }
+ }
+}
+
+object RasterPartitionReader {
+ def rasterToInternalRows(
+ currentRaster: GridCoverage2D,
+ dataSchema: StructType,
+ rasterOptions: RasterOptions,
+ path: Path): Iterator[InternalRow] = {
+ val retile = rasterOptions.retile
+ val tileWidth = rasterOptions.tileWidth
+ val tileHeight = rasterOptions.tileHeight
+ val padWithNoData = rasterOptions.padWithNoData
+
+ val writer = new UnsafeRowWriter(dataSchema.length)
+ writer.resetRowWriter()
+
+ // Extract the file name from the path
+ val fileName = path.getName
+
+ if (retile) {
+ val (tw, th) = (tileWidth, tileHeight) match {
+ case (Some(tw), Some(th)) => (tw, th)
+ case (None, None) =>
+ // Use the internal tile size of the input raster
+ val tw = currentRaster.getRenderedImage.getTileWidth
+ val th = currentRaster.getRenderedImage.getTileHeight
+ val tileSizeError = {
+ """To resolve this issue, you can try one of the following methods:
+ | 1. Disable retile by setting `.option("retile", "false")`.
+ | 2. Explicitly set `tileWidth` and `tileHeight`.
+ | 3. Convert the raster to a Cloud Optimized GeoTIFF (COG) using tools like `gdal_translate`.
+ |""".stripMargin
+ }
+ if (tw >= MAX_AUTO_TILE_SIZE || th >= MAX_AUTO_TILE_SIZE) {
+ throw new IllegalArgumentException(
+ s"Internal tile size of $path is too large ($tw x $th). " + tileSizeError)
+ }
+ if (tw == 0 || th == 0) {
+ throw new IllegalArgumentException(
+ s"Internal tile size of $path contains zero ($tw x $th). " + tileSizeError)
+ }
+ if (tw.toDouble / th > 10.0 || th.toDouble / tw > 10.0) {
+ throw new IllegalArgumentException(
+ s"Internal tile shape of $path is too thin ($tw x $th). " + tileSizeError)
+ }
+ (tw, th)
+ case _ =>
+ throw new IllegalArgumentException("Both tileWidth and tileHeight must be set")
+ }
+
+ val iter =
+ RasterConstructors.generateTiles(currentRaster, null, tw, th, padWithNoData, Double.NaN)
+ iter.asScala.map { tile =>
+ val tileRaster = tile.getCoverage
+ writer.reset()
+ writeRaster(writer, dataSchema, tileRaster, tile.getTileX, tile.getTileY, fileName)
+ tileRaster.dispose(true)
+ writer.getRow
+ }
+ } else {
+ writeRaster(writer, dataSchema, currentRaster, 0, 0, fileName)
+ Iterator.single(writer.getRow)
Review Comment:
`UnsafeRowWriter.getRow` returns an `UnsafeRow` backed by the writer's internal buffer. Since the same writer instance is reused for every tile, returning `writer.getRow` directly can corrupt previously emitted rows when the next tile overwrites the buffer. Return a stable row per record (e.g., copy the row or allocate a new row) before yielding it from the iterator.
```suggestion
writer.getRow.copy()
}
} else {
writeRaster(writer, dataSchema, currentRaster, 0, 0, fileName)
Iterator.single(writer.getRow.copy())
```
##########
spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/raster/RasterPartitionReader.scala:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.spark.sql.sedona_sql.io.raster
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.sedona.common.raster.RasterConstructors
+import org.apache.sedona.common.raster.inputstream.HadoopImageInputStream
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter
+import org.apache.spark.sql.connector.read.PartitionReader
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.sedona_sql.UDT.RasterUDT
+import org.apache.spark.sql.sedona_sql.io.raster.RasterPartitionReader.rasterToInternalRows
+import org.apache.spark.sql.sedona_sql.io.raster.RasterTable.{MAX_AUTO_TILE_SIZE, RASTER, RASTER_NAME, TILE_X, TILE_Y}
+import org.apache.spark.sql.types.StructType
+import org.geotools.coverage.grid.GridCoverage2D
+
+import java.net.URI
+import scala.collection.JavaConverters._
+
+class RasterPartitionReader(
+ configuration: Configuration,
+ partitionedFiles: Array[PartitionedFile],
+ dataSchema: StructType,
+ rasterOptions: RasterOptions)
+ extends PartitionReader[InternalRow] {
+
+ // Track the current file index we're processing
+ private var currentFileIndex = 0
+
+ // Current raster being processed
+ private var currentRaster: GridCoverage2D = _
+
+ // Current image input stream (must be kept open while the raster is in use)
+ private var currentImageStream: HadoopImageInputStream = _
+
+ // Current row
+ private var currentRow: InternalRow = _
+
+ // Iterator for the current file's tiles
+ private var currentIterator: Iterator[InternalRow] = Iterator.empty
+
+ override def next(): Boolean = {
+ // If current iterator has more elements, return true
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+
+ // If current iterator is exhausted, but we have more files, load the next file
+ if (currentFileIndex < partitionedFiles.length) {
+ loadNextFile()
+ if (currentIterator.hasNext) {
+ currentRow = currentIterator.next()
+ return true
+ }
+ }
+
+ // No more data
+ false
+ }
+
+ override def get(): InternalRow = {
+ currentRow
+ }
+
+ override def close(): Unit = {
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+ }
+
+ private def loadNextFile(): Unit = {
+ // Clean up previous raster and stream if exists
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+
+ if (currentFileIndex >= partitionedFiles.length) {
+ currentIterator = Iterator.empty
+ return
+ }
+
+ val partition = partitionedFiles(currentFileIndex)
+ val path = new Path(new URI(partition.filePath.toString()))
+
+ try {
+ // Open a stream-based reader instead of materializing the entire file as byte[].
+ // This avoids the 2 GB byte[] limit and reduces memory pressure for large files.
+ currentImageStream = new HadoopImageInputStream(path, configuration)
+
+ // Create in-db GridCoverage2D from GeoTiff stream. The RenderedImage is lazy -
+ // pixel data will only be decoded when accessed via image.getData(Rectangle).
+ currentRaster = RasterConstructors.fromGeoTiff(currentImageStream)
+ currentIterator = rasterToInternalRows(currentRaster, dataSchema, rasterOptions, path)
+ currentFileIndex += 1
+ } catch {
+ case e: Exception =>
+ if (currentRaster != null) {
+ currentRaster.dispose(true)
+ currentRaster = null
+ }
+ if (currentImageStream != null) {
+ currentImageStream.close()
+ currentImageStream = null
+ }
+ throw e
+ }
+ }
+}
+
+object RasterPartitionReader {
+ def rasterToInternalRows(
+ currentRaster: GridCoverage2D,
+ dataSchema: StructType,
+ rasterOptions: RasterOptions,
+ path: Path): Iterator[InternalRow] = {
+ val retile = rasterOptions.retile
+ val tileWidth = rasterOptions.tileWidth
+ val tileHeight = rasterOptions.tileHeight
+ val padWithNoData = rasterOptions.padWithNoData
+
+ val writer = new UnsafeRowWriter(dataSchema.length)
+ writer.resetRowWriter()
+
+ // Extract the file name from the path
+ val fileName = path.getName
+
+ if (retile) {
+ val (tw, th) = (tileWidth, tileHeight) match {
+ case (Some(tw), Some(th)) => (tw, th)
+ case (None, None) =>
+ // Use the internal tile size of the input raster
+ val tw = currentRaster.getRenderedImage.getTileWidth
+ val th = currentRaster.getRenderedImage.getTileHeight
+ val tileSizeError = {
+ """To resolve this issue, you can try one of the following methods:
+ | 1. Disable retile by setting `.option("retile", "false")`.
+ | 2. Explicitly set `tileWidth` and `tileHeight`.
+ | 3. Convert the raster to a Cloud Optimized GeoTIFF (COG) using tools like `gdal_translate`.
+ |""".stripMargin
+ }
+ if (tw >= MAX_AUTO_TILE_SIZE || th >= MAX_AUTO_TILE_SIZE) {
+ throw new IllegalArgumentException(
+ s"Internal tile size of $path is too large ($tw x $th). " + tileSizeError)
+ }
+ if (tw == 0 || th == 0) {
+ throw new IllegalArgumentException(
+ s"Internal tile size of $path contains zero ($tw x $th). " + tileSizeError)
+ }
+ if (tw.toDouble / th > 10.0 || th.toDouble / tw > 10.0) {
+ throw new IllegalArgumentException(
+ s"Internal tile shape of $path is too thin ($tw x $th). " + tileSizeError)
+ }
+ (tw, th)
+ case _ =>
+ throw new IllegalArgumentException("Both tileWidth and tileHeight must be set")
+ }
+
+ val iter =
+ RasterConstructors.generateTiles(currentRaster, null, tw, th, padWithNoData, Double.NaN)
+ iter.asScala.map { tile =>
+ val tileRaster = tile.getCoverage
+ writer.reset()
+ writeRaster(writer, dataSchema, tileRaster, tile.getTileX, tile.getTileY, fileName)
+ tileRaster.dispose(true)
+ writer.getRow
+ }
+ } else {
+ writeRaster(writer, dataSchema, currentRaster, 0, 0, fileName)
+ Iterator.single(writer.getRow)
Review Comment:
Same buffer-reuse issue for the non-retiled path: `Iterator.single(writer.getRow)` returns an `UnsafeRow` backed by the writer's mutable buffer. Ensure the returned row is not mutated after emission (e.g., return a copy).
```suggestion
val row = writer.getRow.copy()
Iterator.single(row)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org