GEOMESA-3435. run cassandra test for cassandra and scylladb images. add scylladb support to docs

mkorolyov · mkorolyov · commit 07b0818f5120 · 2025-03-08T16:35:19.000+01:00
diff --git a/.github/workflows/build-and-test.yml b/.github/workflows/build-and-test.yml
@@ -19,7 +19,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-      - uses: actions/setup-java@7a6d8a8234af8eb26422e24e3006232cccaa061b # v4.6.0
+      - uses: actions/setup-java@3a4f6e1af504cf6a31855fa899c6aa5355ba6c12 # v4.7.0
         with:
           distribution: 'temurin'
           java-version: '11'
@@ -38,7 +38,7 @@ jobs:
           mv ~/.m2/repository/org/locationtech/geomesa m2-geomesa
           tar -czf classes.tgz $(find . -path '*/target/classes' -o -path '*/target/test-classes' -type d | tr '\n' ' ')
       - name: Upload artifacts
-        uses: actions/upload-artifact@6f51ac03b9356f520e9adb1b1b7802705f340c2b # v4.5.0
+        uses: actions/upload-artifact@65c4c4a1ddee5b72f698fdd19549f0f0fb45cf08 # v4.6.0
         with:
           name: classes-and-jars-${{ inputs.scala-version }}
           retention-days: 1
diff --git a/.github/workflows/codacy.yml b/.github/workflows/codacy.yml
@@ -72,7 +72,7 @@ jobs:
 
       # Upload the SARIF file generated in the previous step
       - name: Upload SARIF results file
-        uses: github/codeql-action/upload-sarif@48ab28a6f5dbc2a99bf1e0131198dd8f1df78169 # v3.28.0
+        uses: github/codeql-action/upload-sarif@dd746615b3b9d728a6a37ca2045b68ca76d4841a # v3.28.8
         with:
           sarif_file: results.sarif
           category: codacy-${{ matrix.tool }}
diff --git a/.github/workflows/dash.yml b/.github/workflows/dash.yml
@@ -19,7 +19,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-      - uses: actions/setup-java@7a6d8a8234af8eb26422e24e3006232cccaa061b # v4.6.0
+      - uses: actions/setup-java@3a4f6e1af504cf6a31855fa899c6aa5355ba6c12 # v4.7.0
         with:
           distribution: 'temurin'
           java-version: '11'
diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml
@@ -19,7 +19,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-      - uses: actions/setup-java@7a6d8a8234af8eb26422e24e3006232cccaa061b # v4.6.0
+      - uses: actions/setup-java@3a4f6e1af504cf6a31855fa899c6aa5355ba6c12 # v4.7.0
         with:
           distribution: 'temurin'
           java-version: '11'
diff --git a/.github/workflows/javadocs.yml b/.github/workflows/javadocs.yml
@@ -19,7 +19,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-      - uses: actions/setup-java@7a6d8a8234af8eb26422e24e3006232cccaa061b # v4.6.0
+      - uses: actions/setup-java@3a4f6e1af504cf6a31855fa899c6aa5355ba6c12 # v4.7.0
         with:
           distribution: 'temurin'
           java-version: '11'
diff --git a/.github/workflows/main-build.yml b/.github/workflows/main-build.yml
@@ -28,15 +28,15 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-      - uses: actions/setup-java@7a6d8a8234af8eb26422e24e3006232cccaa061b # v4.6.0
+      - uses: actions/setup-java@3a4f6e1af504cf6a31855fa899c6aa5355ba6c12 # v4.7.0
         with:
           distribution: 'temurin'
           java-version: '11'
       - uses: actions/cache@1bd1e32a3bdc45362d1e726936510720a7c30a57 # v4.2.0
         with:
           key: ${{ hashFiles('**/pom.xml') }}-docs
           path: ~/.m2/repository/
-      - uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b # v5.3.0
+      - uses: actions/setup-python@42375524e23c412d93fb67b49958b491fce71c38 # v5.4.0
         with:
           python-version: '3.10'
           cache: 'pip'
@@ -52,7 +52,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-      - uses: actions/setup-java@7a6d8a8234af8eb26422e24e3006232cccaa061b # v4.6.0
+      - uses: actions/setup-java@3a4f6e1af504cf6a31855fa899c6aa5355ba6c12 # v4.7.0
         with:
           distribution: 'temurin'
           java-version: '11'
diff --git a/.github/workflows/scorecard.yml b/.github/workflows/scorecard.yml
@@ -59,14 +59,14 @@ jobs:
       # Upload the results as artifacts (optional). Commenting out will disable uploads of run results in SARIF
       # format to the repository Actions tab.
       - name: "Upload artifact"
-        uses: actions/upload-artifact@6f51ac03b9356f520e9adb1b1b7802705f340c2b # v4.5.0
+        uses: actions/upload-artifact@65c4c4a1ddee5b72f698fdd19549f0f0fb45cf08 # v4.6.0
         with:
           name: SARIF file
           path: results.sarif
           retention-days: 5
 
       # Upload the results to GitHub's code scanning dashboard.
       - name: "Upload to code-scanning"
-        uses: github/codeql-action/upload-sarif@48ab28a6f5dbc2a99bf1e0131198dd8f1df78169 # v3.28.0
+        uses: github/codeql-action/upload-sarif@dd746615b3b9d728a6a37ca2045b68ca76d4841a # v3.28.8
         with:
           sarif_file: results.sarif
diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml
@@ -22,7 +22,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-      - uses: actions/setup-java@7a6d8a8234af8eb26422e24e3006232cccaa061b # v4.6.0
+      - uses: actions/setup-java@3a4f6e1af504cf6a31855fa899c6aa5355ba6c12 # v4.7.0
         with:
           distribution: 'temurin'
           java-version: '11'
diff --git a/docs/common.py b/docs/common.py
@@ -99,6 +99,8 @@
 
 .. |cassandra_version| replace:: 3
 
+.. |scylladb_version| replace:: 6.2
+
 .. |redis_supported_versions| replace:: versions 5.0 and later
 
 .. |spark_required_version| replace:: %(spark_version)s
diff --git a/docs/user/index.rst b/docs/user/index.rst
@@ -22,6 +22,7 @@ User Manual
    accumulo/index
    hbase/index
    cassandra/index
+   scylladb/index
    postgis/index
    kafka/index
    redis/index
diff --git a/docs/user/install.rst b/docs/user/install.rst
@@ -8,6 +8,7 @@ Installation instructions are available for each data store implementation:
 * :doc:`/user/accumulo/install`
 * :doc:`/user/hbase/install`
 * :doc:`/user/cassandra/install`
+* :doc:`/user/scylladb/install`
 * :doc:`/user/kafka/install`
 * :doc:`/user/redis/install`
 * :doc:`/user/filesystem/install`
diff --git a/docs/user/scylladb/index.rst b/docs/user/scylladb/index.rst
@@ -0,0 +1,20 @@
+ScyllaDB Data Store
+===================
+
+.. note::
+
+    GeoMesa currently supports ScyllaDB version |scylladb_version|.
+
+The GeoMesa ScyllaDB Data Store is an implementation of the GeoTools
+``DataStore`` interface that is backed by `ScyllaDB`_.
+It is found in the ``geomesa-cassandra`` directory of the GeoMesa
+source distribution as ScyllaDB is a drop-in replacement for Apache Cassandra.
+
+Refer to Cassandra Data Store documentation for usage information.
+
+.. _ScyllaDB: https://www.scylladb.com/
+
+.. toctree::
+    :maxdepth: 1
+
+    install
diff --git a/docs/user/scylladb/install.rst b/docs/user/scylladb/install.rst
@@ -0,0 +1,26 @@
+Installing GeoMesa ScyllaDB
+===========================
+
+.. note::
+
+    GeoMesa currently supports ScyllaDB version |scylladb_version|.
+
+.. note::
+
+    The examples below expect a version to be set in the environment:
+
+    .. parsed-literal::
+
+        $ export TAG="|release_version|"
+        $ export VERSION="|scala_binary_version|-${TAG}" # note: |scala_binary_version| is the Scala build version
+
+Connecting to ScyllaDB
+-----------------------
+
+The first step to getting started with ScyllaDB and GeoMesa is to install
+ScyllaDB itself. You can find good directions for downloading and installing
+ScyllaDB online. For example, see Scylla's official `getting started`_ documentation.
+
+.. _getting started: https://opensource.docs.scylladb.com/stable/getting-started/index.html
+
+Once you have ScyllaDB installed follow the same instructions as for the Apache Cassandra Data Store.
diff --git a/geomesa-accumulo/geomesa-accumulo-datastore/src/main/scala/org/locationtech/geomesa/accumulo/data/AccumuloJoinIndexAdapter.scala b/geomesa-accumulo/geomesa-accumulo-datastore/src/main/scala/org/locationtech/geomesa/accumulo/data/AccumuloJoinIndexAdapter.scala
@@ -73,24 +73,33 @@ object AccumuloJoinIndexAdapter {
 
     // TODO seems like this should be using 'schema' here, which may be a reduced version of the indexSft due to col groups
     val indexSft = index.indexSft
-    lazy val sort = hints.getSortFields
-    lazy val max = hints.getMaxFeatures
-    lazy val project = hints.getProjection
+    val transform = hints.getTransformSchema
 
     // for queries that don't require a join, creates a regular batch scan plan
     def plan(
         iters: Seq[IteratorSetting],
         kvsToFeatures: ResultsToFeatures[Entry[Key, Value]],
-        reduce: Option[FeatureReducer]): BatchScanPlan =
-      BatchScanPlan(filter, tables, ranges, iters, colFamily, kvsToFeatures, reduce, sort, max, project, numThreads)
-
-    val transform = hints.getTransformSchema
+        reduce: Option[FeatureReducer]): BatchScanPlan = {
+      // add the attribute-level vis iterator if necessary
+      val iterators = visibilityIter(index) ++ iters
+      val sort = hints.getSortFields
+      val max = hints.getMaxFeatures
+      val project = hints.getProjection
+      BatchScanPlan(filter, tables, ranges, iterators, colFamily, kvsToFeatures, reduce, sort, max, project, numThreads)
+    }
 
     // used when remote processing is disabled
-    lazy val returnSchema = hints.getTransformSchema.getOrElse(indexSft)
-    lazy val fti = visibilityIter(index) ++ FilterTransformIterator.configure(indexSft, index, ecql, hints).toSeq
-    lazy val resultsToFeatures = AccumuloResultsToFeatures(index, returnSchema)
-    lazy val localReducer = Some(new LocalTransformReducer(returnSchema, None, None, None, hints))
+    def localPlan(overrides: Option[Seq[IteratorSetting]] = None): BatchScanPlan = {
+      val returnSchema = transform.getOrElse(indexSft)
+      if (hints.isSkipReduce) {
+        // override the return sft to reflect what we're actually returning,
+        // since the bin sft is only created in the local reduce step
+        hints.put(QueryHints.Internal.RETURN_SFT, returnSchema)
+      }
+      val iters = overrides.getOrElse(FilterTransformIterator.configure(indexSft, index, ecql, hints).toSeq)
+      val localReducer = Some(new LocalTransformReducer(returnSchema, None, None, None, hints))
+      plan(iters, AccumuloResultsToFeatures(index, returnSchema), localReducer)
+    }
 
     val qp = if (hints.isBinQuery) {
       // check to see if we can execute against the index values
@@ -99,16 +108,10 @@ object AccumuloJoinIndexAdapter {
           hints.getBinLabelField.forall(indexSft.indexOf(_) != -1) &&
           index.supportsFilter(ecql)) {
         if (ds.config.remote.bin) {
-          val iter = BinAggregatingIterator.configure(indexSft, index, ecql, hints)
-          val iters = visibilityIter(index) :+ iter
-          plan(iters, new AccumuloBinResultsToFeatures(), None)
+          val binIter = BinAggregatingIterator.configure(indexSft, index, ecql, hints)
+          plan(Seq(binIter), new AccumuloBinResultsToFeatures(), None)
         } else {
-          if (hints.isSkipReduce) {
-            // override the return sft to reflect what we're actually returning,
-            // since the bin sft is only created in the local reduce step
-            hints.hints.put(QueryHints.Internal.RETURN_SFT, returnSchema)
-          }
-          plan(fti, resultsToFeatures, localReducer)
+          localPlan()
         }
       } else {
         // have to do a join against the record table
@@ -117,40 +120,30 @@ object AccumuloJoinIndexAdapter {
     } else if (hints.isArrowQuery) {
       // check to see if we can execute against the index values
       if (index.canUseIndexSchema(ecql, transform)) {
-        if (ds.config.remote.bin) {
-          val (iter, reduce) = ArrowIterator.configure(indexSft, index, ds.stats, filter.filter, ecql, hints)
-          val iters = visibilityIter(index) :+ iter
-          plan(iters, new AccumuloArrowResultsToFeatures(), Some(reduce))
+        if (ds.config.remote.arrow) {
+          val (arrowIter, reduce) = ArrowIterator.configure(indexSft, index, ds.stats, filter.filter, ecql, hints)
+          plan(Seq(arrowIter), new AccumuloArrowResultsToFeatures(), Some(reduce))
         } else {
-          if (hints.isSkipReduce) {
-            // override the return sft to reflect what we're actually returning,
-            // since the arrow sft is only created in the local reduce step
-            hints.hints.put(QueryHints.Internal.RETURN_SFT, returnSchema)
-          }
-          plan(fti, resultsToFeatures, localReducer)
+          localPlan()
         }
       } else if (index.canUseIndexSchemaPlusKey(ecql, transform)) {
         val transformSft = transform.getOrElse {
           throw new IllegalStateException("Must have a transform for attribute key plus value scan")
         }
         // first filter and apply the transform
         val filterTransformIter = FilterTransformIterator.configure(indexSft, index, ecql, hints, 23).get
-        // clear the transforms as we've already accounted for them
-        hints.clearTransforms()
         // next add the attribute value from the row key
         val rowValueIter = AttributeKeyValueIterator.configure(index.asInstanceOf[AttributeIndex], transformSft, 24)
-        if (ds.config.remote.bin) {
+        if (ds.config.remote.arrow) {
+          // clear the transforms as we've already accounted for them
+          val newHints = new Hints(hints)
+          newHints.clearTransforms()
           // finally apply the arrow iterator on the resulting features
-          val (iter, reduce) = ArrowIterator.configure(transformSft, index, ds.stats, None, None, hints)
-          val iters = visibilityIter(index) ++ Seq(filterTransformIter, rowValueIter, iter)
+          val (arrowIter, reduce) = ArrowIterator.configure(transformSft, index, ds.stats, None, None, newHints)
+          val iters = Seq(filterTransformIter, rowValueIter, arrowIter)
           plan(iters, new AccumuloArrowResultsToFeatures(), Some(reduce))
         } else {
-          if (hints.isSkipReduce) {
-            // override the return sft to reflect what we're actually returning,
-            // since the arrow sft is only created in the local reduce step
-            hints.hints.put(QueryHints.Internal.RETURN_SFT, returnSchema)
-          }
-          plan(fti, resultsToFeatures, localReducer)
+          localPlan(Some(Seq(filterTransformIter, rowValueIter)))
         }
       } else {
         // have to do a join against the record table
@@ -160,8 +153,7 @@ object AccumuloJoinIndexAdapter {
       // check to see if we can execute against the index values
       val weightIsAttribute = hints.getDensityWeight.contains(index.attributes.head)
       if (index.supportsFilter(ecql) && (weightIsAttribute || hints.getDensityWeight.forall(indexSft.indexOf(_) != -1))) {
-        if (ds.config.remote.bin) {
-          val visIter = visibilityIter(index)
+        if (ds.config.remote.density) {
           val iters = if (weightIsAttribute) {
             // create a transform sft with the attribute added
             val transform = {
@@ -181,18 +173,13 @@ object AccumuloJoinIndexAdapter {
             // priority needs to be between vis iter (21) and density iter (25)
             val keyValueIter = AttributeKeyValueIterator.configure(index.asInstanceOf[AttributeIndex], transform, 23)
             val densityIter = DensityIterator.configure(transform, index, ecql, hints)
-            visIter :+ keyValueIter :+ densityIter
+            Seq(keyValueIter, densityIter)
           } else {
-            visIter :+ DensityIterator.configure(indexSft, index, ecql, hints)
+            Seq(DensityIterator.configure(indexSft, index, ecql, hints))
           }
           plan(iters, new AccumuloDensityResultsToFeatures(), None)
         } else {
-          if (hints.isSkipReduce) {
-            // override the return sft to reflect what we're actually returning,
-            // since the density sft is only created in the local reduce step
-            hints.hints.put(QueryHints.Internal.RETURN_SFT, returnSchema)
-          }
-          plan(fti, resultsToFeatures, localReducer)
+          localPlan()
         }
       } else {
         // have to do a join against the record table
@@ -201,18 +188,12 @@ object AccumuloJoinIndexAdapter {
     } else if (hints.isStatsQuery) {
       // check to see if we can execute against the index values
       if (Try(Stat(indexSft, hints.getStatsQuery)).isSuccess && index.supportsFilter(ecql)) {
-        if (ds.config.remote.bin) {
-          val iter = StatsIterator.configure(indexSft, index, ecql, hints)
-          val iters = visibilityIter(index) :+ iter
+        if (ds.config.remote.stats) {
+          val statsIter = StatsIterator.configure(indexSft, index, ecql, hints)
           val reduce = Some(StatsScan.StatsReducer(indexSft, hints))
-          plan(iters, new AccumuloStatsResultsToFeatures(), reduce)
+          plan(Seq(statsIter), new AccumuloStatsResultsToFeatures(), reduce)
         } else {
-          if (hints.isSkipReduce) {
-            // override the return sft to reflect what we're actually returning,
-            // since the stats sft is only created in the local reduce step
-            hints.hints.put(QueryHints.Internal.RETURN_SFT, returnSchema)
-          }
-          plan(fti, resultsToFeatures, localReducer)
+          localPlan()
         }
       } else {
         // have to do a join against the record table
@@ -225,22 +206,17 @@ object AccumuloJoinIndexAdapter {
       val transformSft = transform.getOrElse {
         throw new IllegalStateException("Must have a transform for attribute value scan")
       }
-      val iter = FilterTransformIterator.configure(indexSft, index, ecql, hints.getTransform, hints.getSampling)
-      // add the attribute-level vis iterator if necessary
-      val iters = visibilityIter(index) ++ iter.toSeq
+      val fti = FilterTransformIterator.configure(indexSft, index, ecql, hints.getTransform, hints.getSampling)
       // need to use transform to convert key/values
       val toFeatures = AccumuloResultsToFeatures(index, transformSft)
-      plan(iters, toFeatures, None)
+      plan(fti.toSeq, toFeatures, None)
     } else if (index.canUseIndexSchemaPlusKey(ecql, transform)) {
       // we can use the index PLUS the value
       val transformSft = transform.getOrElse {
         throw new IllegalStateException("Must have a transform for attribute key plus value scan")
       }
-      val iter = FilterTransformIterator.configure(indexSft, index, ecql, hints.getTransform, hints.getSampling)
-      // add the attribute-level vis iterator if necessary
-      val iters =
-        visibilityIter(index) ++ iter.toSeq :+
-            AttributeKeyValueIterator.configure(index.asInstanceOf[AttributeIndex], transformSft)
+      val fti = FilterTransformIterator.configure(indexSft, index, ecql, hints.getTransform, hints.getSampling)
+      val iters = fti.toSeq :+ AttributeKeyValueIterator.configure(index.asInstanceOf[AttributeIndex], transformSft)
       // need to use transform to convert key/values
       val toFeatures = AccumuloResultsToFeatures(index, transformSft)
       plan(iters, toFeatures, None)
@@ -301,7 +277,7 @@ object AccumuloJoinIndexAdapter {
     if (hints.isSkipReduce) {
       // override the return sft to reflect what we're actually returning,
       // since the arrow sft is only created in the local reduce step
-      hints.hints.put(QueryHints.Internal.RETURN_SFT, resultSft)
+      hints.put(QueryHints.Internal.RETURN_SFT, resultSft)
     }
 
     val recordTables = recordIndex.getTablesForQuery(filter.filter)
diff --git a/geomesa-cassandra/geomesa-cassandra-datastore/pom.xml b/geomesa-cassandra/geomesa-cassandra-datastore/pom.xml
diff --git a/geomesa-cassandra/geomesa-cassandra-datastore/src/test/scala/org/locationtech/geomesa/cassandra/data/CassandraDataStoreTest.scala b/geomesa-cassandra/geomesa-cassandra-datastore/src/test/scala/org/locationtech/geomesa/cassandra/data/CassandraDataStoreTest.scala
diff --git a/geomesa-convert/geomesa-convert-common/src/main/scala/org/locationtech/geomesa/convert/EvaluationContext.scala b/geomesa-convert/geomesa-convert-common/src/main/scala/org/locationtech/geomesa/convert/EvaluationContext.scala
diff --git a/geomesa-fs/geomesa-fs-storage/geomesa-fs-storage-convert/src/main/scala/org/locationtech/geomesa/fs/storage/converter/ConverterFileSystemReader.scala b/geomesa-fs/geomesa-fs-storage/geomesa-fs-storage-convert/src/main/scala/org/locationtech/geomesa/fs/storage/converter/ConverterFileSystemReader.scala
diff --git a/pom.xml b/pom.xml
