You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by fr...@apache.org on 2026/05/28 21:42:49 UTC
(cassandra-analytics) branch trunk updated: CASSANALYTICS-34: Spark 4.0 Support (#163)
This is an automated email from the ASF dual-hosted git repository.
frankgh pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-analytics.git
The following commit(s) were added to refs/heads/trunk by this push:
new 1fc500da CASSANALYTICS-34: Spark 4.0 Support (#163)
1fc500da is described below
commit 1fc500da4706a58b619803246c1e7655452b2b25
Author: Francisco Guerrero <fr...@apache.org>
AuthorDate: Thu May 28 14:42:44 2026 -0700
CASSANALYTICS-34: Spark 4.0 Support (#163)
Patch by Francisco Guerrero; reviewed by Yifan Cai for CASSANALYTICS-34
---
.circleci/config.yml | 299 ++++++++++-----------
.github/workflows/test.yaml | 169 +++++++++---
CHANGES.txt | 1 +
analytics-sidecar-client-common/build.gradle | 5 +-
.../sidecar/common/data/RestoreJobStatus.java | 1 +
.../common/request/JsonResponseBytesDecoder.java | 7 +-
analytics-sidecar-client/build.gradle | 5 +-
.../cassandra/sidecar/client/RequestExecutor.java | 53 ++--
.../cassandra/sidecar/client/StreamBuffer.java | 77 ++++++
.../sidecar/client/SidecarClientTest.java | 13 +-
analytics-sidecar-vertx-client-shaded/build.gradle | 24 +-
analytics-sidecar-vertx-client/build.gradle | 5 +-
build.gradle | 33 +--
cassandra-analytics-cdc-codec/build.gradle | 4 +
.../cassandra/cdc/avro/AvroLogicalTypesTest.java | 9 +-
cassandra-analytics-cdc/build.gradle | 3 +
.../java/org/apache/cassandra/cdc/CdcBuilder.java | 1 -
.../apache/cassandra/spark/utils/BuildInfo.java | 27 +-
cassandra-analytics-core-example/build.gradle | 20 +-
cassandra-analytics-core/build.gradle | 31 +--
.../cassandra/spark/bulkwriter/BulkSparkConf.java | 24 +-
.../cassandra/spark/bulkwriter/StreamSession.java | 4 +-
.../org/apache/cassandra/spark/data/DataLayer.java | 19 +-
.../cassandra/spark/data/LocalDataLayer.java | 35 +--
.../cassandra/spark/data/PartitionedDataLayer.java | 46 ++--
.../spark/sparksql/CassandraDataSource.java | 55 ++++
.../spark/sparksql/CassandraInputPartition.java | 33 +--
.../sparksql/CassandraPartitionReaderFactory.java | 69 +++++
.../spark/sparksql/CassandraPartitioning.java | 35 ++-
.../spark/sparksql/CassandraScanBuilder.java | 149 ++++++++++
.../cassandra/spark/sparksql/CassandraTable.java | 68 +++++
.../spark/sparksql/CassandraTableProvider.java | 61 +++++
.../cassandra/spark/sparksql/LocalDataSource.java | 41 ++-
.../spark/sparksql/LocalPartitionSizeSource.java | 37 ++-
.../spark/sparksql/PartitionSizeIterator.java | 96 +++++++
.../spark/sparksql/PartitionSizeTableProvider.java | 171 ++++++++++++
.../cassandra/spark/sparksql/SparkRowIterator.java | 87 ++++++
.../apache/cassandra/spark/SSTableReaderTests.java | 7 +-
.../spark/bulkwriter/BulkSparkConfTest.java | 7 +-
.../spark/bulkwriter/DirectStreamSessionTest.java | 2 +-
.../spark/bulkwriter/RecordWriterTest.java | 4 +-
.../spark/bulkwriter/SortedSSTableWriterTest.java | 2 +-
.../spark/bulkwriter/TokenRangeMappingUtils.java | 2 +-
.../cassandra/spark/data/ClientConfigTests.java | 2 +-
.../cassandra/spark/endtoend/SchemaTests.java | 24 +-
.../spark/utils/BufferingInputStreamTests.java | 88 +++---
.../cassandra/spark/utils/BuildInfoTest.java | 56 ++--
.../apache/cassandra/spark/PartitionSizeTests.java | 95 +++++++
.../spark/bulkwriter/TestTaskContext.java | 218 +++++++++++++++
.../spark/common/SidecarInstanceFactoryTest.java | 55 ++++
cassandra-analytics-integration-tests/build.gradle | 27 +-
.../analytics/BulkWriteDataTypesTest.java | 19 +-
.../apache/cassandra/analytics/SparkTestUtils.java | 8 +-
cassandra-analytics-sidecar-client/build.gradle | 7 +-
.../spark/data/converter/types/StringFeatures.java | 2 +-
.../spark/utils/ScalaConversionUtils.java | 56 ----
.../cassandra/spark/utils/SparkTypeUtils.java | 55 ----
.../spark/utils/ScalaConversionUtils.java | 14 +-
.../cassandra/spark/utils/SparkTypeUtils.java | 11 +-
cassandra-five-zero-bridge/build.gradle | 26 +-
.../cassandra/spark/reader/IndexOffsetTests.java | 12 +-
.../cassandra/spark/reader/SSTableReaderTests.java | 5 +-
.../cassandra/spark/reader/IndexOffsetTests.java | 12 +-
.../cassandra/spark/reader/SSTableReaderTests.java | 5 +-
gradle.properties | 6 +-
gradle/common/java11Options.gradle | 36 +++
gradle/common/java17Options.gradle | 37 +++
gradle/common/javaOptions.gradle | 26 ++
gradle/wrapper/gradle-wrapper.properties | 2 +-
.../scala-2.13-spark-4-jdk-17.gradle | 32 +--
70 files changed, 2000 insertions(+), 777 deletions(-)
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 40a17230..2b38d046 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -49,9 +49,6 @@ commands:
type: string
jdk:
type: string
- use_jdk11:
- type: string
- default: "false"
sstable_format:
type: string
default: "big"
@@ -69,7 +66,6 @@ commands:
INTEGRATION_MAX_HEAP_SIZE: "3072m"
CORE_MAX_PARALLEL_FORKS: 3
CORE_TEST_MAX_HEAP_SIZE: "2048m"
- CASSANDRA_USE_JDK11: <<parameters.use_jdk11>>
CASSANDRA_VERSION: "<<parameters.cassandra>>"
command: |
export GRADLE_OPTS="-Xmx2g -Dorg.gradle.jvmargs=-Xmx2g"
@@ -84,9 +80,6 @@ commands:
type: string
jdk:
type: string
- use_jdk11:
- type: string
- default: "false"
cassandra:
type: string
description: Build and integration test against Spark <<parameters.spark>> Scala <<parameters.scala>> <<parameters.jdk>>
@@ -98,7 +91,6 @@ commands:
JDK_VERSION: "<<parameters.jdk>>"
INTEGRATION_MAX_PARALLEL_FORKS: 1
INTEGRATION_MAX_HEAP_SIZE: "3072M"
- CASSANDRA_USE_JDK11: <<parameters.use_jdk11>>
command: |
export GRADLE_OPTS="-Xmx2g -Dorg.gradle.jvmargs=-Xmx2g"
export DTEST_JAR="dtest-<< parameters.cassandra >>.jar"
@@ -173,9 +165,15 @@ jobs:
- install_common
- checkout
- run:
- name: Build dependencies for jdk11 builds
+ name: Build dependencies for JDK11 builds
+ environment:
+ CASSANDRA_USE_JDK11: "true"
+ # Set JDK_VERSION, SPARK_VERSION, and SCALA_VERSION for codeCheckTasks
+ JDK_VERSION: "11"
+ SPARK_VERSION: "3"
+ SCALA_VERSION: "2.12"
command: |
- CASSANDRA_USE_JDK11=true ./scripts/build-dependencies.sh
+ ./scripts/build-dependencies.sh
./gradlew --no-daemon --max-workers=2 codeCheckTasks
- persist_to_workspace:
root: dependencies
@@ -183,40 +181,54 @@ jobs:
- "*.jar"
- "org/**/*"
- spark3-2_12-jdk11-big-c40:
+ # JDK17 dependency build for the Spark 4 / Scala 2.13 matrix.
+ # Workspace paths overlap with build-deps-jdk11, so any downstream job must
+ # `requires:` only ONE of these two — never both — to avoid jar collisions.
+ build-deps-jdk17:
docker:
- - image: cimg/openjdk:11.0
+ - image: cimg/openjdk:17.0
resource_class: large
steps:
- install_common
- checkout
- - attach_workspace:
- at: dependencies
- - run_build:
- spark: "3"
- scala: "2.12"
- jdk: "11"
- use_jdk11: "true"
- sstable_format: "big"
- cassandra: "4.0"
-
- - store_artifacts:
- when: always
- path: build/test-reports
- destination: test-reports
-
- - store_artifacts:
- when: always
- path: build/reports
- destination: reports
-
- - store_test_results:
- when: always
- path: build/test-reports
+ - run:
+ name: Build dependencies for JDK17 builds
+ environment:
+ # Set JDK_VERSION, SPARK_VERSION, and SCALA_VERSION for codeCheckTasks
+ JDK_VERSION: "17"
+ SPARK_VERSION: "4"
+ SCALA_VERSION: "2.13"
+ # JDK17 only targets Cassandra 5.0+; skip 4.0 / 4.1 dtest jar builds
+ # (build-dtest-jars.sh reads this var to filter CANDIDATE_BRANCHES).
+ BRANCHES: "cassandra-5.0"
+ command: |
+ ./scripts/build-dependencies.sh
+ ./gradlew --no-daemon --max-workers=2 codeCheckTasks
+ - persist_to_workspace:
+ root: dependencies
+ paths:
+ - "*.jar"
+ - "org/**/*"
- spark3-2_12-jdk11-big-c41:
+ # Single parameterized unit-test job. The workflow invokes it via `matrix:`
+ # multiple times, once per "compatibility group" (e.g. C4.x + Scala 2.12 + BIG,
+ # C5.0 + Scala 2.13 + BTI). Splitting into several small matrices is cleaner
+ # than one big matrix with a long `exclude:` list, because CircleCI's
+ # `exclude:` requires every parameter to be listed in every entry.
+ unit:
+ parameters:
+ spark:
+ type: string
+ scala:
+ type: string
+ jdk:
+ type: string
+ cassandra:
+ type: string
+ sstable_format:
+ type: string
docker:
- - image: cimg/openjdk:11.0
+ - image: cimg/openjdk:<<parameters.jdk>>.0
resource_class: large
steps:
- install_common
@@ -224,12 +236,11 @@ jobs:
- attach_workspace:
at: dependencies
- run_build:
- spark: "3"
- scala: "2.12"
- jdk: "11"
- use_jdk11: "true"
- sstable_format: "big"
- cassandra: "4.1"
+ spark: "<<parameters.spark>>"
+ scala: "<<parameters.scala>>"
+ jdk: "<<parameters.jdk>>"
+ sstable_format: "<<parameters.sstable_format>>"
+ cassandra: "<<parameters.cassandra>>"
- store_artifacts:
when: always
@@ -245,47 +256,21 @@ jobs:
when: always
path: build/test-reports
- int-c4-spark3-2_12-jdk11:
- parallelism: 8
- docker:
- - image: cimg/openjdk:11.0
- resource_class: large
- steps:
- - setup_remote_docker
- - install_common
- - checkout
- - attach_workspace:
- at: dependencies
- - run_integration:
- spark: "3"
- scala: "2.12"
- jdk: "11"
- use_jdk11: "true"
- cassandra: "4.0.17"
-
- - store_artifacts:
- when: always
- path: build/test-reports
- destination: test-reports
-
- - store_artifacts:
- when: always
- path: build/reports
- destination: reports
-
- - store_artifacts:
- when: always
- path: cassandra-analytics-integration-tests
- destination: int-tests-misc
-
- - store_test_results:
- when: always
- path: build/aggregated-test-reports
-
- int-c41-spark3-2_12-jdk11:
+ # Single parameterized integration-test job, invoked once per Cassandra/Scala
+ # compatibility group from the workflow's `matrix:` block.
+ int-test:
+ parameters:
+ spark:
+ type: string
+ scala:
+ type: string
+ jdk:
+ type: string
+ cassandra:
+ type: string
parallelism: 8
docker:
- - image: cimg/openjdk:11.0
+ - image: cimg/openjdk:<<parameters.jdk>>.0
resource_class: large
steps:
- setup_remote_docker
@@ -294,11 +279,10 @@ jobs:
- attach_workspace:
at: dependencies
- run_integration:
- spark: "3"
- scala: "2.12"
- jdk: "11"
- use_jdk11: "true"
- cassandra: "4.1.4"
+ spark: "<<parameters.spark>>"
+ scala: "<<parameters.scala>>"
+ jdk: "<<parameters.jdk>>"
+ cassandra: "<<parameters.cassandra>>"
- store_artifacts:
when: always
@@ -319,93 +303,86 @@ jobs:
when: always
path: build/aggregated-test-reports
- spark3-2_13-jdk11-bti-c50:
- docker:
- - image: cimg/openjdk:11.0
- resource_class: large
- steps:
- - install_common
- - checkout
- - attach_workspace:
- at: dependencies
- - run_build:
- spark: "3"
- scala: "2.13"
- jdk: "11"
- use_jdk11: "true"
- sstable_format: "bti"
- cassandra: "5.0"
-
- - store_artifacts:
- when: always
- path: build/test-reports
- destination: test-reports
-
- - store_artifacts:
- when: always
- path: build/reports
- destination: reports
-
- - store_test_results:
- when: always
- path: build/test-reports
-
- int-c5-spark3-2_13-jdk11:
- parallelism: 8
- docker:
- - image: cimg/openjdk:11.0
- resource_class: large
- steps:
- - setup_remote_docker
- - install_common
- - checkout
- - attach_workspace:
- at: dependencies
- - run_integration:
- spark: "3"
- scala: "2.13"
- jdk: "11"
- use_jdk11: "true"
- cassandra: "5.0.5"
-
- - store_artifacts:
- when: always
- path: build/test-reports
- destination: test-reports
-
- - store_artifacts:
- when: always
- path: build/reports
- destination: reports
-
- - store_test_results:
- when: always
- path: build/aggregated-test-reports
-
workflows:
version: 2
build-and-test:
jobs:
- build-deps-jdk11
+ - build-deps-jdk17
- # Unit tests: split by Cassandra version to reduce per-process memory pressure
- - spark3-2_12-jdk11-big-c40:
+ # ---- Unit tests ----
+ # Cassandra 4.x: Scala 2.12 + BIG sstable format
+ - unit:
requires:
- build-deps-jdk11
- - spark3-2_12-jdk11-big-c41:
+ matrix:
+ alias: unit-c4x
+ parameters:
+ spark: ["3"]
+ scala: ["2.12"]
+ jdk: ["11"]
+ sstable_format: ["big"]
+ cassandra: ["4.0", "4.1"]
+
+ # Cassandra 5.0: Scala 2.13 + BTI sstable format
+ - unit:
requires:
- build-deps-jdk11
- - spark3-2_13-jdk11-bti-c50:
+ matrix:
+ alias: unit-c5
+ parameters:
+ spark: ["3"]
+ scala: ["2.13"]
+ jdk: ["11"]
+ sstable_format: ["bti"]
+ cassandra: ["5.0"]
+
+ # Cassandra 5.0 on Spark 4 / Scala 2.13 / JDK 17 + BTI sstable format
+ - unit:
requires:
- - build-deps-jdk11
-
- # Integration tests
- - int-c4-spark3-2_12-jdk11:
+ - build-deps-jdk17
+ matrix:
+ alias: unit-c5-spark4
+ parameters:
+ spark: ["4"]
+ scala: ["2.13"]
+ jdk: ["17"]
+ sstable_format: ["bti"]
+ cassandra: ["5.0"]
+
+ # ---- Integration tests ----
+ # Cassandra 4.x: Scala 2.12, full patch versions for dtest jar lookup
+ - int-test:
requires:
- build-deps-jdk11
- - int-c41-spark3-2_12-jdk11:
+ matrix:
+ alias: int-c4x
+ parameters:
+ spark: ["3"]
+ scala: ["2.12"]
+ jdk: ["11"]
+ cassandra: ["4.0.17", "4.1.4"]
+
+ # Cassandra 5.0: Scala 2.13
+ - int-test:
requires:
- build-deps-jdk11
- - int-c5-spark3-2_13-jdk11:
+ matrix:
+ alias: int-c5
+ parameters:
+ spark: ["3"]
+ scala: ["2.13"]
+ jdk: ["11"]
+ cassandra: ["5.0.5"]
+
+ # Cassandra 5.0 on Spark 4 / Scala 2.13 / JDK 17
+ - int-test:
requires:
- - build-deps-jdk11
+ - build-deps-jdk17
+ matrix:
+ alias: int-c5-spark4
+ parameters:
+ spark: ["4"]
+ scala: ["2.13"]
+ jdk: ["17"]
+ cassandra: ["5.0.5"]
diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml
index cc662901..0a136c08 100644
--- a/.github/workflows/test.yaml
+++ b/.github/workflows/test.yaml
@@ -26,8 +26,8 @@ on:
workflow_dispatch:
jobs:
- build:
- name: Compile and build
+ build-jdk11:
+ name: Compile and build (JDK 11)
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
@@ -38,43 +38,107 @@ jobs:
java-version: 11
- run: |
sudo apt-get update
-
+
+ if [ -f /etc/ssl/certs/java/cacerts/cacerts ]; then
+ sudo mv /etc/ssl/certs/java/cacerts/ /etc/ssl/certs/java/cacerts-old
+ sudo mv /etc/ssl/certs/java/cacerts-old/cacerts /etc/ssl/certs/java/
+ sudo rmdir /etc/ssl/certs/java/cacerts-old
+ fi
+
+ apt-get download ant ant-optional
+ sudo dpkg --force-all -i ant*.deb
+ rm ant*.deb
+
+ sudo bash -c 'for i in {2..20}; do echo 127.0.0.${i} localhost${i} >> /etc/hosts; done'
+
+ for i in {2..20}
+ do
+ sudo ip addr add "127.0.0.${i}" dev lo
+ sudo route add -host "127.0.0.${i}" dev lo;
+ done
+
+ export JDK_VERSION="11"
+ export SPARK_VERSION="3"
+ export SCALA_VERSION="2.12"
+ export CASSANDRA_USE_JDK11="true"
+
+ ./scripts/build-dependencies.sh
+
+ ./gradlew --no-daemon --max-workers=2 codeCheckTasks
+ - name: Cache Maven repository
+ uses: actions/cache@v4
+ with:
+ path: ~/.m2
+ key: maven-repo-jdk11-${{ github.sha }}
+ - name: Cache workspace
+ id: cache-build-save
+ uses: actions/cache/save@v4
+ with:
+ path: ${{ github.workspace }}
+ key: build-jdk11-${{ github.sha }}
+
+ # JDK17 dependency build for the Spark 4 / Scala 2.13 / Cassandra 5.0 matrix.
+ # Produces a distinct workspace cache (build-jdk17-...) so downstream test jobs
+ # must restore from exactly one of build-jdk11 / build-jdk17 — mixing them
+ # would clobber dependency jars.
+ build-jdk17:
+ name: Compile and build (JDK 17)
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v4
+ - name: Setup JDK
+ uses: actions/setup-java@v4
+ with:
+ distribution: 'temurin'
+ java-version: 17
+ - run: |
+ sudo apt-get update
+
if [ -f /etc/ssl/certs/java/cacerts/cacerts ]; then
sudo mv /etc/ssl/certs/java/cacerts/ /etc/ssl/certs/java/cacerts-old
sudo mv /etc/ssl/certs/java/cacerts-old/cacerts /etc/ssl/certs/java/
sudo rmdir /etc/ssl/certs/java/cacerts-old
fi
-
+
apt-get download ant ant-optional
sudo dpkg --force-all -i ant*.deb
rm ant*.deb
-
+
sudo bash -c 'for i in {2..20}; do echo 127.0.0.${i} localhost${i} >> /etc/hosts; done'
-
+
for i in {2..20}
do
sudo ip addr add "127.0.0.${i}" dev lo
sudo route add -host "127.0.0.${i}" dev lo;
done
-
- CASSANDRA_USE_JDK11=true ./scripts/build-dependencies.sh
-
+
+ export JDK_VERSION="17"
+ export SPARK_VERSION="4"
+ export SCALA_VERSION="2.13"
+ # JDK17 only targets Cassandra 5.0+; skip 4.0 / 4.1 dtest jar builds
+ # (build-dtest-jars.sh reads this var to filter CANDIDATE_BRANCHES).
+ export BRANCHES="cassandra-5.0"
+
+ ./scripts/build-dependencies.sh
+
./gradlew codeCheckTasks
- name: Cache Maven repository
uses: actions/cache@v4
with:
path: ~/.m2
- key: maven-repo-${{ github.sha }}
+ key: maven-repo-jdk17-${{ github.sha }}
- name: Cache workspace
id: cache-build-save
uses: actions/cache/save@v4
with:
path: ${{ github.workspace }}
- key: build-${{ github.sha }}
+ key: build-jdk17-${{ github.sha }}
unit-test:
- name: Unit test - Scala ${{ matrix.scala }} ${{ matrix.sstable-format }} C${{ matrix.cassandra }}
- needs: build
+ name: Unit test - Scala ${{ matrix.scala }} ${{ matrix.sstable-format }} C${{ matrix.cassandra }} Spark${{ matrix.spark }} JDK${{ matrix.jdk }}
+ # Each matrix entry only consumes one of these caches (selected by matrix.jdk),
+ # but `needs:` cannot be matrix-conditional so we wait on both builds.
+ needs: [build-jdk11, build-jdk17]
runs-on: ubuntu-latest
strategy:
matrix:
@@ -82,19 +146,30 @@ jobs:
- scala: '2.13'
sstable-format: 'bti'
cassandra: '5.0'
+ jdk: '11'
+ spark: '3'
- scala: '2.12'
sstable-format: 'big'
cassandra: '4.1'
+ jdk: '11'
+ spark: '3'
- scala: '2.12'
sstable-format: 'big'
cassandra: '4.0'
+ jdk: '11'
+ spark: '3'
+ - scala: '2.13'
+ sstable-format: 'bti'
+ cassandra: '5.0'
+ jdk: '17'
+ spark: '4'
fail-fast: false
steps:
- name: Setup JDK
uses: actions/setup-java@v4
with:
distribution: 'temurin'
- java-version: 11
+ java-version: ${{ matrix.jdk }}
- run: |
sudo bash -c 'for i in {2..20}; do echo 127.0.0.${i} localhost${i} >> /etc/hosts; done'
for i in {2..20}
@@ -106,54 +181,70 @@ jobs:
uses: actions/cache@v4
with:
path: ~/.m2
- key: maven-repo-${{ github.sha }}
+ key: maven-repo-jdk${{ matrix.jdk }}-${{ github.sha }}
- name: Cache workspace
id: cache-build-restore
uses: actions/cache/restore@v4
with:
path: ${{ github.workspace }}
- key: build-${{ github.sha }}
+ key: build-jdk${{ matrix.jdk }}-${{ github.sha }}
- run: |
- export SPARK_VERSION="3"
+ export SPARK_VERSION="${{ matrix.spark }}"
export SCALA_VERSION="${{ matrix.scala }}"
- export JDK_VERSION="11"
+ export JDK_VERSION="${{ matrix.jdk }}"
export INTEGRATION_MAX_PARALLEL_FORKS=1
export INTEGRATION_MAX_HEAP_SIZE="3072M"
- export CASSANDRA_USE_JDK11=true
export CASSANDRA_VERSION="${{ matrix.cassandra }}"
+ if [ "${{ matrix.jdk }}" = "11" ]; then
+ export CASSANDRA_USE_JDK11=true
+ fi
./gradlew --stacktrace clean assemble check -x cassandra-analytics-integration-tests:test -Dcassandra.analytics.bridges.sstable_format=${{ matrix.sstable-format }}
integration-test:
name: Integration test - ${{ matrix.config }} (${{ matrix.job_index }})
- needs: build
+ # Each matrix entry only consumes one of these caches (selected by matrix.jdk),
+ # but `needs:` cannot be matrix-conditional so we wait on both builds.
+ needs: [build-jdk11, build-jdk17]
runs-on: ubuntu-latest
strategy:
- # GitHub Actions generate a cross-product of 'config' × 'job_index' (3 × 5 = 15 jobs).
+ # GitHub Actions generate a cross-product of 'config' × 'job_index' (4 × 5 = 20 jobs).
# The 'include' entries don't add new combinations — they augment existing ones
- # by matching on 'config' and injecting 'scala' and 'cassandra' into each match.
- # To add a new version: add one entry to 'config' and one to 'include'.
+ # by matching on 'config' and injecting 'scala', 'cassandra', 'jdk', and 'spark'
+ # into each match. To add a new version: add one entry to 'config' and one to
+ # 'include'.
matrix:
- config: ['s2.13-c5.0.5', 's2.12-c4.1.4', 's2.12-c4.0.17']
+ config: ['s2.13-c5.0.5', 's2.12-c4.1.4', 's2.12-c4.0.17', 's2.13-c5.0.5-spark4']
job_index: [0, 1, 2, 3, 4]
job_total: [5]
include:
- config: 's2.13-c5.0.5'
scala: '2.13'
cassandra: '5.0.5'
+ jdk: '11'
+ spark: '3'
- config: 's2.12-c4.1.4'
scala: '2.12'
cassandra: '4.1.4'
+ jdk: '11'
+ spark: '3'
- config: 's2.12-c4.0.17'
scala: '2.12'
cassandra: '4.0.17'
+ jdk: '11'
+ spark: '3'
+ - config: 's2.13-c5.0.5-spark4'
+ scala: '2.13'
+ cassandra: '5.0.5'
+ jdk: '17'
+ spark: '4'
fail-fast: false
steps:
- name: Setup JDK
uses: actions/setup-java@v4
with:
distribution: 'temurin'
- java-version: 11
+ java-version: ${{ matrix.jdk }}
- run: |
sudo bash -c 'for i in {2..20}; do echo 127.0.0.${i} localhost${i} >> /etc/hosts; done'
for i in {2..20}
@@ -165,26 +256,28 @@ jobs:
uses: actions/cache@v4
with:
path: ~/.m2
- key: maven-repo-${{ github.sha }}
+ key: maven-repo-jdk${{ matrix.jdk }}-${{ github.sha }}
- name: Cache workspace
id: cache-build-restore
uses: actions/cache/restore@v4
with:
path: ${{ github.workspace }}
- key: build-${{ github.sha }}
+ key: build-jdk${{ matrix.jdk }}-${{ github.sha }}
- run: |
- export SPARK_VERSION="3"
+ export SPARK_VERSION="${{ matrix.spark }}"
export SCALA_VERSION="${{ matrix.scala }}"
- export JDK_VERSION="11"
+ export JDK_VERSION="${{ matrix.jdk }}"
export INTEGRATION_MAX_PARALLEL_FORKS=1
export INTEGRATION_MAX_HEAP_SIZE="3072M"
- export CASSANDRA_USE_JDK11=true
-
+ if [ "${{ matrix.jdk }}" = "11" ]; then
+ export CASSANDRA_USE_JDK11=true
+ fi
+
export DTEST_JAR="dtest-${{ matrix.cassandra }}.jar"
export CASSANDRA_VERSION=$(echo ${{ matrix.cassandra }} | cut -d'.' -f 1,2)
-
+
./gradlew --stacktrace clean assemble
-
+
cd cassandra-analytics-integration-tests/src/test/java
# Shuffle test classes using the commit SHA as seed for reproducible randomization,
# then shard across runners via round-robin on the shuffled order.
@@ -192,7 +285,7 @@ jobs:
| python3 -c "import random,sys; lines=sys.stdin.read().splitlines(); random.seed('$GITHUB_SHA'); random.shuffle(lines); print('\n'.join(lines))" \
| awk 'NR % ${{ matrix.job_total }} == ${{ matrix.job_index }}')
cd ../../../..
-
+
EXIT_STATUS=0
# Execution of "gradle test --test $TEST_NAME" returns non-zero exit code when commend did not run any test
# (e.g. when all tests are ignored). Currently there is no option to change Gradle behaviour.
@@ -219,13 +312,13 @@ jobs:
echo "Skipping test $TEST_NAME"
fi
done;
-
+
tar czf integration-tests.tar.gz test-reports/*
-
+
exit $EXIT_STATUS
- name: Publish test results
uses: actions/upload-artifact@v4
if: (!cancelled())
with:
- name: integration-tests-${{ matrix.scala }}-${{ matrix.cassandra }}-${{ matrix.job_index }}
- path: integration-tests.tar.gz
\ No newline at end of file
+ name: integration-tests-${{ matrix.config }}-${{ matrix.job_index }}
+ path: integration-tests.tar.gz
diff --git a/CHANGES.txt b/CHANGES.txt
index 4e5f27ed..17e9b858 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * Spark 4.0 Support (CASSANALYTICS-34)
* Make BulkWriterConfig extensible (CASSANALYTICS-168)
0.4.0
diff --git a/analytics-sidecar-client-common/build.gradle b/analytics-sidecar-client-common/build.gradle
index e742f65c..2133ef46 100644
--- a/analytics-sidecar-client-common/build.gradle
+++ b/analytics-sidecar-client-common/build.gradle
@@ -27,13 +27,10 @@ plugins {
id 'maven-publish'
}
-if (propertyWithDefault("artifactType", null) == "common")
-{
+if (propertyWithDefault("artifactType", null) == "common") {
apply from: "$rootDir/gradle/common/publishing.gradle"
}
-sourceCompatibility = 1.8
-
test {
useJUnitPlatform()
maxParallelForks = Runtime.runtime.availableProcessors().intdiv(2) ?: 1
diff --git a/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobStatus.java b/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobStatus.java
index 9074ccb5..7ecf9c7f 100644
--- a/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobStatus.java
+++ b/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobStatus.java
@@ -45,6 +45,7 @@ public enum RestoreJobStatus
/**
* @deprecated replaced by {@link #ABORTED}
*/
+ @Deprecated
FAILED,
/**
* The external controller aborts the RestoreJob due to some failure, e.g. consistency not satisfied, timeout, etc.
diff --git a/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/request/JsonResponseBytesDecoder.java b/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/request/JsonResponseBytesDecoder.java
index 20929641..6367baa2 100644
--- a/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/request/JsonResponseBytesDecoder.java
+++ b/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/request/JsonResponseBytesDecoder.java
@@ -25,13 +25,14 @@ import com.fasterxml.jackson.databind.ObjectMapper;
/**
* Decoder for json response body bytes
+ *
* @param <T> expected java type
*/
public class JsonResponseBytesDecoder<T> implements ResponseBytesDecoder<T>
{
private static final ObjectMapper MAPPER = new ObjectMapper()
- // ignore all the properties that are not declared
- .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ // ignore all the properties that are not declared
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
private final Class<T> type;
public JsonResponseBytesDecoder(Class<T> type)
@@ -42,7 +43,7 @@ public class JsonResponseBytesDecoder<T> implements ResponseBytesDecoder<T>
@Override
public T decode(byte[] bytes) throws IOException
{
- if (bytes == null)
+ if (bytes == null || bytes.length == 0)
{
return null;
}
diff --git a/analytics-sidecar-client/build.gradle b/analytics-sidecar-client/build.gradle
index 202c592d..37c67cd0 100644
--- a/analytics-sidecar-client/build.gradle
+++ b/analytics-sidecar-client/build.gradle
@@ -29,13 +29,10 @@ plugins {
id('maven-publish')
}
-if (propertyWithDefault("artifactType", null) == "common")
-{
+if (propertyWithDefault("artifactType", null) == "common") {
apply from: "$rootDir/gradle/common/publishing.gradle"
}
-sourceCompatibility = 1.8
-
test {
useJUnitPlatform()
testLogging {
diff --git a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java
index 15f92aeb..eb535cef 100644
--- a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java
+++ b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java
@@ -47,7 +47,7 @@ public class RequestExecutor implements AutoCloseable
protected final HttpClient httpClient;
protected final ScheduledExecutorService singleThreadExecutorService;
- protected RequestExecutor(HttpClient httpClient)
+ public RequestExecutor(HttpClient httpClient)
{
this.httpClient = requireNonNull(httpClient, "The httpClient is required");
this.singleThreadExecutorService = Executors.newSingleThreadScheduledExecutor();
@@ -159,6 +159,13 @@ public class RequestExecutor implements AutoCloseable
public void close() throws Exception
{
httpClient.close();
+
+ singleThreadExecutorService.shutdown();
+ if (!singleThreadExecutorService.awaitTermination(1, TimeUnit.MINUTES))
+ {
+ logger.warn("Executor service did not terminate within 1 minute");
+ singleThreadExecutorService.shutdownNow();
+ }
}
/**
@@ -267,17 +274,17 @@ public class RequestExecutor implements AutoCloseable
Request request = context.request();
context.retryPolicy()
.onResponse(future, request, response, throwable, attempt, retryOnNewHost, (nextAttempt, delay) -> {
- String statusCode = response != null ? String.valueOf(response.statusCode()) : "<Not Available>";
- SidecarInstance nextInstance = iterator.hasNext() ? iterator.next() : sidecarInstance;
- if (response == null || response.statusCode() != HttpResponseStatus.ACCEPTED.code())
- {
- logger.warn("Retrying request on {} instance after {}ms. " +
- "Failed on instance={}, attempt={}, statusCode={}",
- nextInstance == sidecarInstance ? "same" : "next", delay,
- sidecarInstance, attempt, statusCode, throwable);
- }
- schedule(delay, () -> executeWithRetries(future, iterator, nextInstance, context, nextAttempt));
- });
+ String statusCode = response != null ? String.valueOf(response.statusCode()) : "<Not Available>";
+ SidecarInstance nextInstance = iterator.hasNext() ? iterator.next() : sidecarInstance;
+ if (response == null || response.statusCode() != HttpResponseStatus.ACCEPTED.code())
+ {
+ logger.warn("Retrying request on {} instance after {}ms. " +
+ "Failed on instance={}, attempt={}, statusCode={}",
+ nextInstance == sidecarInstance ? "same" : "next", delay,
+ sidecarInstance, attempt, statusCode, throwable);
+ }
+ schedule(delay, () -> executeWithRetries(future, iterator, nextInstance, context, nextAttempt));
+ });
}
/**
@@ -306,17 +313,17 @@ public class RequestExecutor implements AutoCloseable
Request request = context.request();
context.retryPolicy()
.onResponse(future, request, response, throwable, attempt, retryOnNewHost, (nextAttempt, delay) -> {
- String statusCode = response != null ? String.valueOf(response.statusCode()) : "<Not Available>";
- SidecarInstance nextInstance = iterator.hasNext() ? iterator.next() : sidecarInstance;
- if (response == null || response.statusCode() != HttpResponseStatus.ACCEPTED.code())
- {
- logger.warn("Retrying stream on {} instance after {}ms. " +
- "Failed on instance={}, attempt={}, statusCode={}",
- nextInstance == sidecarInstance ? "same" : "next", delay,
- sidecarInstance, attempt, statusCode, throwable);
- }
- schedule(delay, () -> streamWithRetries(future, consumer, iterator, nextInstance, context, nextAttempt));
- });
+ String statusCode = response != null ? String.valueOf(response.statusCode()) : "<Not Available>";
+ SidecarInstance nextInstance = iterator.hasNext() ? iterator.next() : sidecarInstance;
+ if (response == null || response.statusCode() != HttpResponseStatus.ACCEPTED.code())
+ {
+ logger.warn("Retrying stream on {} instance after {}ms. " +
+ "Failed on instance={}, attempt={}, statusCode={}",
+ nextInstance == sidecarInstance ? "same" : "next", delay,
+ sidecarInstance, attempt, statusCode, throwable);
+ }
+ schedule(delay, () -> streamWithRetries(future, consumer, iterator, nextInstance, context, nextAttempt));
+ });
}
/**
diff --git a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/StreamBuffer.java b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/StreamBuffer.java
index 4664c040..353bfafa 100644
--- a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/StreamBuffer.java
+++ b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/StreamBuffer.java
@@ -82,6 +82,17 @@ public interface StreamBuffer
return new ByteArrayWrapper(bytes);
}
+ /**
+ * Wraps a {@link ByteBuffer} into a {@link ByteBufferWrapper}
+ *
+ * @param buffer the underlying ByteBuffer
+ * @return the object wrapping ByteBuffer
+ */
+ static ByteBufferWrapper wrap(ByteBuffer buffer)
+ {
+ return new ByteBufferWrapper(buffer);
+ }
+
/**
* A {@link StreamBuffer} implementation that wraps a byte array
*/
@@ -123,4 +134,70 @@ public interface StreamBuffer
{
}
}
+
+ /**
+ * A {@link StreamBuffer} implementation that wraps a {@link ByteBuffer}.
+ * This implementation assumes single-threaded access and may modify the
+ * buffer's position and limit during operations.
+ */
+ class ByteBufferWrapper implements StreamBuffer
+ {
+ private final ByteBuffer buffer;
+
+ private ByteBufferWrapper(ByteBuffer buffer)
+ {
+ this.buffer = buffer;
+ }
+
+ @Override
+ public void copyBytes(int sourceOffset, ByteBuffer destination, int length)
+ {
+ int originalPosition = buffer.position();
+ int originalLimit = buffer.limit();
+ try
+ {
+ buffer.position(sourceOffset);
+ buffer.limit(sourceOffset + length);
+ destination.put(buffer);
+ destination.flip();
+ }
+ finally
+ {
+ buffer.position(originalPosition);
+ buffer.limit(originalLimit);
+ }
+ }
+
+ @Override
+ public void copyBytes(int sourceOffset, byte[] destination, int destinationIndex, int length)
+ {
+ int originalPosition = buffer.position();
+ try
+ {
+ buffer.position(sourceOffset);
+ buffer.get(destination, destinationIndex, length);
+ }
+ finally
+ {
+ buffer.position(originalPosition);
+ }
+ }
+
+ @Override
+ public byte getByte(int index)
+ {
+ return buffer.get(index);
+ }
+
+ @Override
+ public int readableBytes()
+ {
+ return buffer.remaining();
+ }
+
+ @Override
+ public void release()
+ {
+ }
+ }
}
diff --git a/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java b/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java
index ccf29c3a..3d6a1a1f 100644
--- a/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java
+++ b/analytics-sidecar-client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java
@@ -1303,13 +1303,12 @@ abstract class SidecarClientTest
assertThat(request3.getHeader("User-Agent")).isEqualTo("cassandra-sidecar-test/0.0.1");
assertThat(request3.getHeader("range")).isEqualTo("bytes=10-20");
- byte[] bytes = receivedBytes.stream()
- .collect(ByteArrayOutputStream::new,
- (outputStream, src) -> outputStream.write(src, 0, src.length),
- (outputStream, src) -> {
- })
- .toByteArray();
- assertThat(new String(bytes, StandardCharsets.UTF_8)).isEqualTo("TOC.txt\nSt");
+ String actual = receivedBytes.stream()
+ .collect(ByteArrayOutputStream::new,
+ (outputStream, src) -> outputStream.write(src, 0, src.length),
+ (outputStream, src) -> {
+ }).toString(StandardCharsets.UTF_8);
+ assertThat(actual).isEqualTo("TOC.txt\nSt");
}
}
diff --git a/analytics-sidecar-vertx-client-shaded/build.gradle b/analytics-sidecar-vertx-client-shaded/build.gradle
index 1d1bdeb1..028327ca 100644
--- a/analytics-sidecar-vertx-client-shaded/build.gradle
+++ b/analytics-sidecar-vertx-client-shaded/build.gradle
@@ -33,13 +33,11 @@ plugins {
id('java-library')
id('maven-publish')
id('signing')
- id('com.gradleup.shadow') version '8.3.0'
+ id('com.gradleup.shadow') version '8.3.9'
}
version project.version
-sourceCompatibility = 1.8
-
configurations {
all*.exclude(group: 'ch.qos.logback')
}
@@ -69,6 +67,23 @@ tasks.named('test') {
// Relocating a Package
shadowJar {
archiveClassifier.set('')
+
+ // Exclude unsupported java versions from jar files with multi-release class support
+ exclude 'META-INF/versions/12/'
+ exclude 'META-INF/versions/13/'
+ exclude 'META-INF/versions/14/'
+ exclude 'META-INF/versions/15/'
+ exclude 'META-INF/versions/16/'
+ exclude 'META-INF/versions/17/'
+ exclude 'META-INF/versions/18/'
+ exclude 'META-INF/versions/19/'
+ exclude 'META-INF/versions/20/'
+ exclude 'META-INF/versions/21/'
+ exclude 'META-INF/versions/22/'
+ exclude 'META-INF/versions/23/'
+ exclude 'META-INF/versions/24/'
+ exclude 'META-INF/versions/25/'
+
// Our use of Jackson should be an implementation detail - shade everything so no matter what
// version of Jackson is available in the classpath we don't break consumers of the client
relocate 'org.apache.cassandra.sidecar.common', 'o.a.c.sidecar.client.shaded.common'
@@ -107,8 +122,7 @@ ext {
// analytics-sidecar-vertx-client-shaded_* as the artifact name
archivesBaseName = "analytics-sidecar-vertx-client-all"
-if (propertyWithDefault("artifactType", null) == "common")
-{
+if (propertyWithDefault("artifactType", null) == "common") {
publishing {
publications {
shadow(MavenPublication) { publication ->
diff --git a/analytics-sidecar-vertx-client/build.gradle b/analytics-sidecar-vertx-client/build.gradle
index f04aa732..fa6a0d32 100644
--- a/analytics-sidecar-vertx-client/build.gradle
+++ b/analytics-sidecar-vertx-client/build.gradle
@@ -25,10 +25,7 @@ plugins {
id('maven-publish')
}
-sourceCompatibility = 1.8
-
-if (propertyWithDefault("artifactType", null) == "common")
-{
+if (propertyWithDefault("artifactType", null) == "common") {
apply from: "$rootDir/gradle/common/publishing.gradle"
}
diff --git a/build.gradle b/build.gradle
index 065e903f..258aba16 100644
--- a/build.gradle
+++ b/build.gradle
@@ -109,7 +109,6 @@ apply(from: profile)
apply(plugin: 'idea')
-
// Force checkstyle and rat to run before test tasks for faster feedback
def codeCheckTasks = task("codeCheckTasks")
@@ -190,6 +189,8 @@ subprojects {
apply(plugin: 'java-library')
apply(plugin: 'checkstyle')
+ apply from: "${project.rootDir}/gradle/common/javaOptions.gradle"
+
sourceCompatibility = "${project.rootProject.ext.jdkLabel}"
targetCompatibility = "${project.rootProject.ext.jdkLabel}"
@@ -259,28 +260,14 @@ subprojects {
def heapDumpPath = "${project.rootProject.rootDir}/build/${project.name}/heapDumps"
Files.createDirectories(Paths.get(heapDumpPath))
- if (JavaVersion.current().isJava11Compatible()) {
- def JDK11_OPTIONS = ['-Djdk.attach.allowAttachSelf=true',
- '--add-exports', 'java.base/jdk.internal.misc=ALL-UNNAMED',
- '--add-exports', 'java.base/jdk.internal.ref=ALL-UNNAMED',
- '--add-exports', 'java.base/sun.nio.ch=ALL-UNNAMED',
- '--add-exports', 'java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED',
- '--add-exports', 'java.rmi/sun.rmi.registry=ALL-UNNAMED',
- '--add-exports', 'java.rmi/sun.rmi.server=ALL-UNNAMED',
- '--add-exports', 'java.sql/java.sql=ALL-UNNAMED',
- '--add-opens', 'java.base/java.lang.module=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.loader=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.ref=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.reflect=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.math=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.module=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.util.jar=ALL-UNNAMED',
- '--add-opens', 'jdk.management/com.sun.management.internal=ALL-UNNAMED',
- '-XX:+HeapDumpOnOutOfMemoryError',
- "-XX:HeapDumpPath=${heapDumpPath}"]
- jvmArgs(JDK11_OPTIONS)
- println("JVM arguments for $project.name are $allJvmArgs")
- }
+ def jdkOptions = project.ext.JDK_OPTIONS
+
+ jdkOptions += [
+ '-XX:+HeapDumpOnOutOfMemoryError',
+ "-XX:HeapDumpPath=${heapDumpPath}",
+ ]
+ jvmArgs(jdkOptions)
+ println("JVM arguments for $project.name are $allJvmArgs")
}
codeCheckTasks.dependsOn(tasks.withType(Checkstyle))
diff --git a/cassandra-analytics-cdc-codec/build.gradle b/cassandra-analytics-cdc-codec/build.gradle
index 60b72208..8ed8489b 100644
--- a/cassandra-analytics-cdc-codec/build.gradle
+++ b/cassandra-analytics-cdc-codec/build.gradle
@@ -68,7 +68,11 @@ sourceSets {
}
}
+apply from: "${project.rootDir}/gradle/common/javaOptions.gradle"
+
test {
+ jvmArgs(project.ext.JDK_OPTIONS)
+
useJUnitPlatform()
reports {
def destDir = Paths.get(rootProject.rootDir.absolutePath, "build", "test-reports", "four-zero-types").toFile()
diff --git a/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/AvroLogicalTypesTest.java b/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/AvroLogicalTypesTest.java
index 420f12c3..c55b62bc 100644
--- a/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/AvroLogicalTypesTest.java
+++ b/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/AvroLogicalTypesTest.java
@@ -25,6 +25,7 @@ import java.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalTime;
+import java.time.temporal.ChronoUnit;
import java.util.UUID;
import org.apache.avro.AvroRuntimeException;
@@ -93,13 +94,17 @@ public class AvroLogicalTypesTest
@Test
public void testTimeMicros() throws IOException
{
- testLogicalTypeDataWriteAndReadBack("time_micros", LocalTime.now());
+ // In https://bugs.openjdk.org/browse/JDK-8242504 the precision of clock was enhanced to
+ // nanoseconds. However, for Cassandra we only need microseconds
+ testLogicalTypeDataWriteAndReadBack("time_micros", LocalTime.now().truncatedTo(ChronoUnit.MICROS));
}
@Test
public void testTimestampMicros() throws IOException
{
- testLogicalTypeDataWriteAndReadBack("timestamp_micros", Instant.now());
+ // In https://bugs.openjdk.org/browse/JDK-8242504 the precision of clock was enhanced to
+ // nanoseconds. However, for Cassandra we only need microseconds
+ testLogicalTypeDataWriteAndReadBack("timestamp_micros", Instant.now().truncatedTo(ChronoUnit.MICROS));
}
private <T> void testLogicalTypeDataWriteAndReadBack(String typeName, T testData) throws IOException
diff --git a/cassandra-analytics-cdc/build.gradle b/cassandra-analytics-cdc/build.gradle
index e413b242..b335ebe8 100644
--- a/cassandra-analytics-cdc/build.gradle
+++ b/cassandra-analytics-cdc/build.gradle
@@ -141,6 +141,8 @@ jar {
}
}
+apply from: "${project.rootDir}/gradle/common/javaOptions.gradle"
+
// Assuming container w/8g ram, 2x3G == 6G + overhead; override via CDC_MAX_PARALLEL_FORKS for local dev
def cdcMaxHeap = '3072m'
def cdcMaxParallelForks = (System.getenv("CDC_MAX_PARALLEL_FORKS") ?: "2") as int
@@ -164,6 +166,7 @@ def configureCdcTestTask = { Test task, String majorMinor = null ->
task.maxHeapSize = cdcMaxHeap
task.maxParallelForks = cdcMaxParallelForks
task.forkEvery = 1 // Enables different end-to-end test classes use Spark contexts with different configurations
+ task.jvmArgs(project.ext.JDK_OPTIONS)
// Make it so unit tests run on a Jar with Cassandra bridge implementations built in
task.dependsOn(tasks.jar)
diff --git a/cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/CdcBuilder.java b/cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/CdcBuilder.java
index d95315ed..cd8b3ea9 100644
--- a/cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/CdcBuilder.java
+++ b/cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/CdcBuilder.java
@@ -119,7 +119,6 @@ public class CdcBuilder
return withExecutor(AsyncExecutor.wrap(executor));
}
-
public CdcBuilder withExecutor(@NotNull AsyncExecutor asyncExecutor)
{
this.asyncExecutor = asyncExecutor;
diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/BuildInfo.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/BuildInfo.java
index 8e3ffd86..6a4f5a13 100644
--- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/BuildInfo.java
+++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/BuildInfo.java
@@ -66,6 +66,29 @@ public final class BuildInfo
* @return {@code true} if the java version is at least java 11, {@code false} otherwise
*/
public static boolean isAtLeastJava11(String version)
+ {
+ return isAtLeastJavaVersion(version, 11);
+ }
+
+ /**
+ * Determine whether the provided version is at least java 17
+ *
+ * @param version the java specification version
+ * @return {@code true} if the java version is at least java 17, {@code false} otherwise
+ */
+ public static boolean isAtLeastJava17(String version)
+ {
+ return isAtLeastJavaVersion(version, 17);
+ }
+
+ /**
+ * Determine whether the provided version is at least the minimum java version
+ *
+ * @param version the java specification version
+ * @param minimumVersion the minimum version required
+ * @return {@code true} if the java version is at least the minimum java version, {@code false} otherwise
+ */
+ public static boolean isAtLeastJavaVersion(String version, int minimumVersion)
{
if (version == null)
{
@@ -73,11 +96,11 @@ public final class BuildInfo
}
else if (version.contains("."))
{
- return version.compareTo("11") >= 0;
+ return version.compareTo(minimumVersion + "") >= 0;
}
else
{
- return Integer.parseInt(version) >= 11;
+ return Integer.parseInt(version) >= minimumVersion;
}
}
diff --git a/cassandra-analytics-core-example/build.gradle b/cassandra-analytics-core-example/build.gradle
index b8ab3d40..ad017517 100644
--- a/cassandra-analytics-core-example/build.gradle
+++ b/cassandra-analytics-core-example/build.gradle
@@ -33,6 +33,8 @@ dependencies {
implementation(group: "${sparkGroupId}", name: "spark-sql_${scalaMajorVersion}", version: "${project.rootProject.sparkVersion}")
}
+apply from: "${project.rootDir}/gradle/common/javaOptions.gradle"
+
application {
// Optionally allow to start with a different mainClass, rather than the default 'DirectWriteAndReadJob'
// For example,
@@ -40,22 +42,6 @@ application {
// or
// ./gradlew :cassandra-analytics-core-example:run --args='LocalS3WriteAndReadJob'
mainClass = 'org.apache.cassandra.spark.example.JobSelector'
- applicationDefaultJvmArgs = ["-Dfile.encoding=UTF-8",
- "-Djdk.attach.allowAttachSelf=true",
- "--add-exports", "java.base/jdk.internal.misc=ALL-UNNAMED",
- "--add-exports", "java.base/jdk.internal.ref=ALL-UNNAMED",
- "--add-exports", "java.base/sun.nio.ch=ALL-UNNAMED",
- "--add-exports", "java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED",
- "--add-exports", "java.rmi/sun.rmi.registry=ALL-UNNAMED",
- "--add-exports", "java.rmi/sun.rmi.server=ALL-UNNAMED",
- "--add-exports", "java.sql/java.sql=ALL-UNNAMED",
- "--add-opens", "java.base/java.lang.module=ALL-UNNAMED",
- "--add-opens", "java.base/jdk.internal.loader=ALL-UNNAMED",
- "--add-opens", "java.base/jdk.internal.ref=ALL-UNNAMED",
- "--add-opens", "java.base/jdk.internal.reflect=ALL-UNNAMED",
- "--add-opens", "java.base/jdk.internal.math=ALL-UNNAMED",
- "--add-opens", "java.base/jdk.internal.module=ALL-UNNAMED",
- "--add-opens", "java.base/jdk.internal.util.jar=ALL-UNNAMED",
- "--add-opens", "jdk.management/com.sun.management.internal=ALL-UNNAMED"]
+ applicationDefaultJvmArgs = project.ext.JDK_OPTIONS
}
diff --git a/cassandra-analytics-core/build.gradle b/cassandra-analytics-core/build.gradle
index 8df6b139..ca70ba2a 100644
--- a/cassandra-analytics-core/build.gradle
+++ b/cassandra-analytics-core/build.gradle
@@ -25,11 +25,12 @@ plugins {
id('maven-publish')
}
-if (propertyWithDefault("artifactType", null) == "spark")
-{
+if (propertyWithDefault("artifactType", null) == "spark") {
apply from: "$rootDir/gradle/common/publishing.gradle"
}
+apply from: "${project.rootDir}/gradle/common/javaOptions.gradle"
+
java {
withJavadocJar()
withSourcesJar()
@@ -159,26 +160,8 @@ jar {
// (CI) or falls back to all supported versions (local dev).
def configureCoreTestTask(Test task, String majorMinor = null) {
task.doFirst {
- if (JavaVersion.current().isJava11Compatible()) {
- def JDK11_OPTIONS = ['-Djdk.attach.allowAttachSelf=true',
- '--add-exports', 'java.base/jdk.internal.misc=ALL-UNNAMED',
- '--add-exports', 'java.base/jdk.internal.ref=ALL-UNNAMED',
- '--add-exports', 'java.base/sun.nio.ch=ALL-UNNAMED',
- '--add-exports', 'java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED',
- '--add-exports', 'java.rmi/sun.rmi.registry=ALL-UNNAMED',
- '--add-exports', 'java.rmi/sun.rmi.server=ALL-UNNAMED',
- '--add-exports', 'java.sql/java.sql=ALL-UNNAMED',
- '--add-opens', 'java.base/java.lang.module=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.loader=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.ref=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.reflect=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.math=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.module=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.util.jar=ALL-UNNAMED',
- '--add-opens', 'jdk.management/com.sun.management.internal=ALL-UNNAMED']
- task.jvmArgs(JDK11_OPTIONS)
- println("JVM arguments for $project.name are $allJvmArgs")
- }
+ task.jvmArgs(project.ext.JDK_OPTIONS)
+ println("JVM arguments for $project.name are $allJvmArgs")
}
task.systemProperty "cassandra.analytics.bridges.sstable_format", System.getProperty("cassandra.analytics.bridges.sstable_format", "big")
@@ -222,6 +205,8 @@ test {
maxParallelForks = System.getenv('CORE_MAX_PARALLEL_FORKS')?.toInteger()
?: Math.max(Runtime.runtime.availableProcessors() * 2, 8)
+ jvmArgs(project.ext.JDK_OPTIONS)
+
useJUnitPlatform {
excludeTags 'Sequential'
}
@@ -261,7 +246,7 @@ rootProject.ext.cassandraVersionEnumMap.keySet().each { majorMinor ->
/* Start: JaCoCo check */
jacoco {
- toolVersion = '0.8.4'
+ toolVersion = '0.8.13'
}
jacocoTestReport {
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java
index 68f5636b..c50389f7 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java
@@ -75,6 +75,23 @@ public class BulkSparkConf implements Serializable
+ " --add-opens java.base/jdk.internal.util.jar=ALL-UNNAMED"
+ " --add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED";
+ public static final String JDK17_OPTIONS = " --add-exports java.rmi/sun.rmi.transport=ALL-UNNAMED"
+ + " --add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED"
+ + " --add-opens java.base/java.io=ALL-UNNAMED"
+ + " --add-opens java.base/java.lang=ALL-UNNAMED"
+ + " --add-opens java.base/java.lang.invoke=ALL-UNNAMED"
+ + " --add-opens java.base/java.lang.reflect=ALL-UNNAMED"
+ + " --add-opens java.base/java.math=ALL-UNNAMED"
+ + " --add-opens java.base/java.net=ALL-UNNAMED"
+ + " --add-opens java.base/java.nio=ALL-UNNAMED"
+ + " --add-opens java.base/java.util=ALL-UNNAMED"
+ + " --add-opens java.base/java.util.concurrent=ALL-UNNAMED"
+ + " --add-opens java.base/java.util.concurrent.atomic=ALL-UNNAMED"
+ + " --add-opens java.base/jdk.internal.misc=ALL-UNNAMED"
+ + " --add-opens java.base/sun.nio.ch=ALL-UNNAMED"
+ + " --add-opens java.base/sun.util.calendar=ALL-UNNAMED"
+ + " --add-opens java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED";
+
public static final int DEFAULT_NUM_SPLITS = -1;
public static final int DEFAULT_HTTP_CONNECTION_TIMEOUT = 100_000;
public static final int DEFAULT_HTTP_RESPONSE_TIMEOUT = 100_000;
@@ -666,7 +683,12 @@ public class BulkSparkConf implements Serializable
public static void setupSparkConf(SparkConf conf, boolean addKryoRegistrator)
{
String previousOptions = conf.get("spark.executor.extraJavaOptions", "");
- if (BuildInfo.isAtLeastJava11(BuildInfo.javaSpecificationVersion()))
+
+ if (BuildInfo.isAtLeastJava17(BuildInfo.javaSpecificationVersion()))
+ {
+ conf.set("spark.executor.extraJavaOptions", previousOptions + JDK11_OPTIONS + JDK17_OPTIONS);
+ }
+ else if (BuildInfo.isAtLeastJava11(BuildInfo.javaSpecificationVersion()))
{
conf.set("spark.executor.extraJavaOptions", previousOptions + JDK11_OPTIONS);
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java
index 9a05dca6..40cbbdea 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java
@@ -29,6 +29,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicReference;
@@ -41,7 +42,6 @@ import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import o.a.c.sidecar.client.shaded.io.vertx.core.impl.ConcurrentHashSet;
import org.apache.cassandra.bridge.SSTableDescriptor;
import org.apache.cassandra.spark.bulkwriter.token.ReplicaAwareFailureHandler;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
@@ -61,7 +61,7 @@ public abstract class StreamSession<T extends TransportContext>
protected final SortedSSTableWriter sstableWriter;
protected final ExecutorService executorService;
- private final Set<Path> streamedFiles = new ConcurrentHashSet<>();
+ private final Set<Path> streamedFiles = ConcurrentHashMap.newKeySet();
private final AtomicReference<Exception> lastStreamFailure = new AtomicReference<>();
private volatile boolean isStreamFinalized = false;
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/DataLayer.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/DataLayer.java
index 394913ac..bb087dec 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/DataLayer.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/DataLayer.java
@@ -29,8 +29,6 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;
-import org.apache.commons.lang.StringUtils;
-
import org.apache.cassandra.bridge.BigNumberConfig;
import org.apache.cassandra.bridge.CassandraBridge;
import org.apache.cassandra.bridge.CassandraBridgeFactory;
@@ -318,7 +316,7 @@ public abstract class DataLayer implements Serializable
public Filter[] unsupportedPushDownFilters(Filter[] filters)
{
Set<String> partitionKeys = cqlTable().partitionKeys().stream()
- .map(key -> StringUtils.lowerCase(key.name()))
+ .map(key -> lowerCase(key.name()))
.collect(Collectors.toSet());
List<Filter> unsupportedFilters = new ArrayList<>(filters.length);
@@ -326,9 +324,9 @@ public abstract class DataLayer implements Serializable
{
if (filter instanceof EqualTo || filter instanceof In)
{
- String columnName = StringUtils.lowerCase(filter instanceof EqualTo
- ? ((EqualTo) filter).attribute()
- : ((In) filter).attribute());
+ String columnName = lowerCase(filter instanceof EqualTo
+ ? ((EqualTo) filter).attribute()
+ : ((In) filter).attribute());
if (partitionKeys.contains(columnName))
{
@@ -359,4 +357,13 @@ public abstract class DataLayer implements Serializable
{
return Stats.DoNothingStats.INSTANCE;
}
+
+ static String lowerCase(String s)
+ {
+ if (s == null)
+ {
+ return s;
+ }
+ return s.toLowerCase();
+ }
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/LocalDataLayer.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/LocalDataLayer.java
index a6d44904..64b57e1c 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/LocalDataLayer.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/LocalDataLayer.java
@@ -35,6 +35,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
@@ -43,9 +44,6 @@ import java.util.stream.Stream;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
-import org.apache.commons.lang.StringUtils;
-import org.apache.commons.lang.builder.EqualsBuilder;
-import org.apache.commons.lang.builder.HashCodeBuilder;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
@@ -179,7 +177,7 @@ public class LocalDataLayer extends DataLayer implements Serializable
getOrThrow(options, lowerCaseKey("keyspace")),
getOrThrow(options, lowerCaseKey("createStmt")),
Arrays.stream(options.getOrDefault(lowerCaseKey("udts"), "").split("\n"))
- .filter(StringUtils::isNotEmpty)
+ .filter(s -> !s.isEmpty())
.collect(Collectors.toSet()),
SchemaFeatureSet.initializeFromOptions(options),
getBoolean(options, lowerCaseKey("useBufferingInputStream"), getBoolean(options, lowerCaseKey("useSSTableInputStream"), false)),
@@ -407,35 +405,20 @@ public class LocalDataLayer extends DataLayer implements Serializable
@Override
public int hashCode()
{
- return new HashCodeBuilder()
- .append(cqlTable)
- .append(paths)
- .append(version())
- .toHashCode();
+ return Objects.hash(cqlTable, Arrays.hashCode(paths), version());
}
@Override
- public boolean equals(Object other)
+ public boolean equals(Object o)
{
- if (other == null)
+ if (o == null || getClass() != o.getClass())
{
return false;
}
- if (this == other)
- {
- return true;
- }
- if (this.getClass() != other.getClass())
- {
- return false;
- }
-
- LocalDataLayer that = (LocalDataLayer) other;
- return new EqualsBuilder()
- .append(this.cqlTable, that.cqlTable)
- .append(this.paths, that.paths)
- .append(this.version(), that.version())
- .isEquals();
+ LocalDataLayer that = (LocalDataLayer) o;
+ return Objects.equals(cqlTable, that.cqlTable)
+ && Objects.deepEquals(paths, that.paths)
+ && Objects.equals(version(), that.version());
}
// JDK Serialization
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/PartitionedDataLayer.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/PartitionedDataLayer.java
index 72ee3d7c..235ae049 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/PartitionedDataLayer.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/PartitionedDataLayer.java
@@ -28,6 +28,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@@ -38,8 +39,6 @@ import java.util.stream.Stream;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Range;
-import org.apache.commons.lang.builder.EqualsBuilder;
-import org.apache.commons.lang.builder.HashCodeBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -88,7 +87,7 @@ public abstract class PartitionedDataLayer extends DataLayer
}
public static final Comparator<AvailabilityHint> AVAILABILITY_HINT_COMPARATOR =
- Comparator.comparingInt((AvailabilityHint other) -> other.priority);
+ Comparator.comparingInt((AvailabilityHint other) -> other.priority);
public static AvailabilityHint fromState(String status, String state)
{
@@ -209,16 +208,16 @@ public abstract class PartitionedDataLayer extends DataLayer
LOGGER.error("Unable to find the sparkTokenRange for partitionId={} in reversePartitionMap={}",
partitionId, reversePartitionMap);
throw new IllegalStateException(
- String.format("Unable to find sparkTokenRange for partitionId=%d in the reverse partition map",
- partitionId));
+ String.format("Unable to find sparkTokenRange for partitionId=%d in the reverse partition map",
+ partitionId));
}
return SparkRangeFilter.create(RangeUtils.toTokenRange(sparkTokenRange));
}
@Override
public List<PartitionKeyFilter> partitionKeyFiltersInRange(
- int partitionId,
- List<PartitionKeyFilter> filters) throws NoMatchFoundException
+ int partitionId,
+ List<PartitionKeyFilter> filters) throws NoMatchFoundException
{
// We only need to worry about Partition key filters that overlap with this Spark workers token range
SparkRangeFilter rangeFilter = sparkRangeFilter(partitionId);
@@ -380,9 +379,9 @@ public abstract class PartitionedDataLayer extends DataLayer
@NotNull Map<Range<BigInteger>, List<CassandraInstance>> ranges)
{
return ranges.values().stream()
- .flatMap(Collection::stream)
- .filter(instance -> !consistencyLevel.isDCLocal || dataCenter == null || instance.dataCenter().equals(dataCenter))
- .collect(Collectors.toSet());
+ .flatMap(Collection::stream)
+ .filter(instance -> !consistencyLevel.isDCLocal || dataCenter == null || instance.dataCenter().equals(dataCenter))
+ .collect(Collectors.toSet());
}
/**
@@ -442,7 +441,7 @@ public abstract class PartitionedDataLayer extends DataLayer
/**
* Return a set of primary and backup CassandraInstances to satisfy the consistency level.
- *
+ * <p>
* NOTE: This method current assumes that each Spark token worker owns a single replica set.
*
* @param instances replicas that overlap with the Spark worker's token range
@@ -472,7 +471,7 @@ public abstract class PartitionedDataLayer extends DataLayer
// multiple replica sets but for current implementation of the TokenPartitioner
// it returns a single replica set per Spark worker/partition
LOGGER.warn("Cannot use incremental repair awareness when Spark partition owns more than one replica set, "
- + "performance will be degraded numRanges={}", ranges.size());
+ + "performance will be degraded numRanges={}", ranges.size());
replicaSet.incrementalRepairPrimary = null;
}
@@ -547,30 +546,17 @@ public abstract class PartitionedDataLayer extends DataLayer
@Override
public int hashCode()
{
- return new HashCodeBuilder()
- .append(datacenter)
- .toHashCode();
+ return Objects.hashCode(datacenter);
}
@Override
- public boolean equals(Object other)
+ public boolean equals(Object o)
{
- if (other == null)
- {
- return false;
- }
- if (this == other)
- {
- return true;
- }
- if (this.getClass() != other.getClass())
+ if (o == null || getClass() != o.getClass())
{
return false;
}
-
- PartitionedDataLayer that = (PartitionedDataLayer) other;
- return new EqualsBuilder()
- .append(this.datacenter, that.datacenter)
- .isEquals();
+ PartitionedDataLayer that = (PartitionedDataLayer) o;
+ return Objects.equals(datacenter, that.datacenter);
}
}
diff --git a/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraDataSource.java b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraDataSource.java
new file mode 100644
index 00000000..b025b6d8
--- /dev/null
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraDataSource.java
@@ -0,0 +1,55 @@
+/*
+ * 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.cassandra.spark.sparksql;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.cassandra.bridge.CassandraBridgeFactory;
+import org.apache.cassandra.bridge.CassandraVersion;
+import org.apache.cassandra.spark.data.CassandraDataLayer;
+import org.apache.cassandra.spark.data.CassandraDataSourceHelper;
+import org.apache.cassandra.spark.data.ClientConfig;
+import org.apache.cassandra.spark.data.DataLayer;
+
+public class CassandraDataSource extends CassandraTableProvider
+{
+ public CassandraDataSource()
+ {
+ CassandraBridgeFactory.validateBridges(CassandraVersion.implementedVersions());
+ }
+
+ @Override
+ public String shortName()
+ {
+ return "cassandraBulkRead";
+ }
+
+ @Override
+ public DataLayer getDataLayer(org.apache.spark.sql.util.CaseInsensitiveStringMap options)
+ {
+ return CassandraDataSourceHelper.getDataLayer(options, this::initializeDataLayer);
+ }
+
+ @VisibleForTesting
+ protected void initializeDataLayer(CassandraDataLayer dataLayer, ClientConfig config)
+ {
+ dataLayer.initialize(config);
+ }
+}
diff --git a/cassandra-analytics-sidecar-client/build.gradle b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraInputPartition.java
similarity index 56%
copy from cassandra-analytics-sidecar-client/build.gradle
copy to cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraInputPartition.java
index bedf2c04..ea76f9b7 100644
--- a/cassandra-analytics-sidecar-client/build.gradle
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraInputPartition.java
@@ -1,5 +1,3 @@
-import java.nio.file.Paths
-
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -19,26 +17,21 @@ import java.nio.file.Paths
* under the License.
*/
-plugins {
- id('java-library')
- id('maven-publish')
-}
+package org.apache.cassandra.spark.sparksql;
-configurations {
- all*.exclude(group: 'org.slf4j', module: 'slf4j-log4j12')
- all*.exclude(group: 'log4j', module: 'log4j')
-}
+import org.apache.spark.sql.connector.read.InputPartition;
-if (propertyWithDefault("artifactType", null) == "common")
+class CassandraInputPartition implements InputPartition
{
- apply from: "$rootDir/gradle/common/publishing.gradle"
-}
-
-dependencies {
- compileOnly("com.google.guava:guava:${guavaVersion}")
- implementation project(":cassandra-analytics-common")
+ private final int partitionId;
- implementation "org.slf4j:slf4j-api:${slf4jApiVersion}"
+ CassandraInputPartition(int partitionId)
+ {
+ this.partitionId = partitionId;
+ }
- api(project(path: ':analytics-sidecar-vertx-client-shaded', configuration: 'shadow'))
-}
\ No newline at end of file
+ public int getPartitionId()
+ {
+ return partitionId;
+ }
+}
diff --git a/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraPartitionReaderFactory.java b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraPartitionReaderFactory.java
new file mode 100644
index 00000000..b2a7df7a
--- /dev/null
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraPartitionReaderFactory.java
@@ -0,0 +1,69 @@
+/*
+ * 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.cassandra.spark.sparksql;
+
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.spark.data.DataLayer;
+import org.apache.cassandra.spark.sparksql.filters.PartitionKeyFilter;
+import org.apache.spark.TaskContext;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.read.InputPartition;
+import org.apache.spark.sql.connector.read.PartitionReader;
+import org.apache.spark.sql.connector.read.PartitionReaderFactory;
+import org.apache.spark.sql.types.StructType;
+
+class CassandraPartitionReaderFactory implements PartitionReaderFactory
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(CassandraPartitionReaderFactory.class);
+ final DataLayer dataLayer;
+ final StructType requiredSchema;
+ final List<PartitionKeyFilter> partitionKeyFilters;
+
+ CassandraPartitionReaderFactory(DataLayer dataLayer,
+ StructType requiredSchema,
+ List<PartitionKeyFilter> partitionKeyFilters)
+ {
+ this.dataLayer = dataLayer;
+ this.requiredSchema = requiredSchema;
+ this.partitionKeyFilters = partitionKeyFilters;
+ }
+
+ @Override
+ public PartitionReader<InternalRow> createReader(InputPartition partition)
+ {
+ int partitionId;
+ if (partition instanceof CassandraInputPartition)
+ {
+ partitionId = ((CassandraInputPartition) partition).getPartitionId();
+ }
+ else
+ {
+ partitionId = TaskContext.getPartitionId();
+ LOGGER.warn("InputPartition is not of CassandraInputPartition type. "
+ + "Using TaskContext to determine the partitionId type={}, partitionId={}",
+ partition.getClass().getName(), partitionId);
+ }
+ return new SparkRowIterator(partitionId, dataLayer, requiredSchema, partitionKeyFilters);
+ }
+}
diff --git a/cassandra-analytics-sidecar-client/build.gradle b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraPartitioning.java
similarity index 56%
copy from cassandra-analytics-sidecar-client/build.gradle
copy to cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraPartitioning.java
index bedf2c04..e305f365 100644
--- a/cassandra-analytics-sidecar-client/build.gradle
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraPartitioning.java
@@ -1,5 +1,3 @@
-import java.nio.file.Paths
-
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -19,26 +17,23 @@ import java.nio.file.Paths
* under the License.
*/
-plugins {
- id('java-library')
- id('maven-publish')
-}
+package org.apache.cassandra.spark.sparksql;
-configurations {
- all*.exclude(group: 'org.slf4j', module: 'slf4j-log4j12')
- all*.exclude(group: 'log4j', module: 'log4j')
-}
+import org.apache.cassandra.spark.data.DataLayer;
+import org.apache.spark.sql.connector.read.partitioning.Partitioning;
-if (propertyWithDefault("artifactType", null) == "common")
+class CassandraPartitioning implements Partitioning
{
- apply from: "$rootDir/gradle/common/publishing.gradle"
-}
-
-dependencies {
- compileOnly("com.google.guava:guava:${guavaVersion}")
- implementation project(":cassandra-analytics-common")
+ final DataLayer dataLayer;
- implementation "org.slf4j:slf4j-api:${slf4jApiVersion}"
+ CassandraPartitioning(DataLayer dataLayer)
+ {
+ this.dataLayer = dataLayer;
+ }
- api(project(path: ':analytics-sidecar-vertx-client-shaded', configuration: 'shadow'))
-}
\ No newline at end of file
+ @Override
+ public int numPartitions()
+ {
+ return dataLayer.partitionCount();
+ }
+}
diff --git a/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraScanBuilder.java b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraScanBuilder.java
new file mode 100644
index 00000000..2f821398
--- /dev/null
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraScanBuilder.java
@@ -0,0 +1,149 @@
+/*
+ * 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.cassandra.spark.sparksql;
+
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.apache.cassandra.spark.data.CqlField;
+import org.apache.cassandra.spark.data.DataLayer;
+import org.apache.cassandra.spark.sparksql.filters.PartitionKeyFilter;
+import org.apache.cassandra.spark.utils.FilterUtils;
+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.ScanBuilder;
+import org.apache.spark.sql.connector.read.SupportsPushDownFilters;
+import org.apache.spark.sql.connector.read.SupportsPushDownRequiredColumns;
+import org.apache.spark.sql.connector.read.SupportsReportPartitioning;
+import org.apache.spark.sql.connector.read.partitioning.Partitioning;
+import org.apache.spark.sql.sources.Filter;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+class CassandraScanBuilder implements ScanBuilder, Scan, Batch, SupportsPushDownFilters, SupportsPushDownRequiredColumns, SupportsReportPartitioning
+{
+ final DataLayer dataLayer;
+ final StructType schema;
+ final CaseInsensitiveStringMap options;
+ StructType requiredSchema = null;
+ Filter[] pushedFilters = new Filter[0];
+
+ CassandraScanBuilder(DataLayer dataLayer, StructType schema, CaseInsensitiveStringMap options)
+ {
+ this.dataLayer = dataLayer;
+ this.schema = schema;
+ this.options = options;
+ }
+
+ @Override
+ public Scan build()
+ {
+ return this;
+ }
+
+ @Override
+ public void pruneColumns(StructType requiredSchema)
+ {
+ this.requiredSchema = requiredSchema;
+ }
+
+ @Override
+ public Filter[] pushFilters(Filter[] filters)
+ {
+ Filter[] unsupportedFilters = dataLayer.unsupportedPushDownFilters(filters);
+
+ List<Filter> supportedFilters = new ArrayList<>(Arrays.asList(filters));
+ supportedFilters.removeAll(Arrays.asList(unsupportedFilters));
+ pushedFilters = supportedFilters.toArray(new Filter[0]);
+
+ return unsupportedFilters;
+ }
+
+ @Override
+ public Filter[] pushedFilters()
+ {
+ return pushedFilters;
+ }
+
+ @Override
+ public StructType readSchema()
+ {
+ return requiredSchema;
+ }
+
+ @Override
+ public Batch toBatch()
+ {
+ return this;
+ }
+
+ @Override
+ public InputPartition[] planInputPartitions()
+ {
+ return IntStream.range(0, dataLayer.partitionCount())
+ .mapToObj(CassandraInputPartition::new)
+ .toArray(InputPartition[]::new);
+ }
+
+ @Override
+ public PartitionReaderFactory createReaderFactory()
+ {
+ return new CassandraPartitionReaderFactory(dataLayer, requiredSchema, buildPartitionKeyFilters());
+ }
+
+ @Override
+ public Partitioning outputPartitioning()
+ {
+ return new CassandraPartitioning(dataLayer);
+ }
+
+ private List<PartitionKeyFilter> buildPartitionKeyFilters()
+ {
+ List<String> partitionKeyColumnNames = dataLayer.cqlTable().partitionKeys().stream().map(CqlField::name).collect(Collectors.toList());
+ Map<String, List<String>> partitionKeyValues = FilterUtils.extractPartitionKeyValues(pushedFilters, new HashSet<>(partitionKeyColumnNames));
+ if (partitionKeyValues.size() > 0)
+ {
+ List<List<String>> orderedValues = partitionKeyColumnNames.stream().map(partitionKeyValues::get).collect(Collectors.toList());
+ return FilterUtils.cartesianProduct(orderedValues).stream()
+ .map(this::buildFilter)
+ .collect(Collectors.toList());
+ }
+ else
+ {
+ return new ArrayList<>();
+ }
+ }
+
+ private PartitionKeyFilter buildFilter(List<String> keys)
+ {
+ AbstractMap.SimpleEntry<ByteBuffer, BigInteger> filterKey = dataLayer.bridge().getPartitionKey(dataLayer.cqlTable(), dataLayer.partitioner(), keys);
+ return PartitionKeyFilter.create(filterKey.getKey(), filterKey.getValue());
+ }
+}
diff --git a/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraTable.java b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraTable.java
new file mode 100644
index 00000000..35790a96
--- /dev/null
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraTable.java
@@ -0,0 +1,68 @@
+/*
+ * 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.cassandra.spark.sparksql;
+
+import java.util.Set;
+
+import com.google.common.collect.Sets;
+
+import org.apache.cassandra.spark.data.DataLayer;
+import org.apache.spark.sql.connector.catalog.SupportsRead;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableCapability;
+import org.apache.spark.sql.connector.read.ScanBuilder;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+class CassandraTable implements Table, SupportsRead
+{
+ private final DataLayer dataLayer;
+ private final StructType schema;
+
+ CassandraTable(DataLayer dataLayer, StructType schema)
+ {
+ this.dataLayer = dataLayer;
+ this.schema = schema;
+ }
+
+ @Override
+ public String name()
+ {
+ return dataLayer.cqlTable().keyspace() + "." + dataLayer.cqlTable().table();
+ }
+
+ @Override
+ public StructType schema()
+ {
+ return schema;
+ }
+
+ @Override
+ public Set<TableCapability> capabilities()
+ {
+ return Sets.newHashSet(TableCapability.BATCH_READ, TableCapability.MICRO_BATCH_READ);
+ }
+
+ @Override
+ public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options)
+ {
+ return new CassandraScanBuilder(dataLayer, schema, options);
+ }
+}
diff --git a/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraTableProvider.java b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraTableProvider.java
new file mode 100644
index 00000000..a2e86b10
--- /dev/null
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/CassandraTableProvider.java
@@ -0,0 +1,61 @@
+/*
+ * 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.cassandra.spark.sparksql;
+
+import java.util.Map;
+
+import org.apache.cassandra.spark.data.DataLayer;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableProvider;
+import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.sources.DataSourceRegister;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+public abstract class CassandraTableProvider implements TableProvider, DataSourceRegister
+{
+ private DataLayer dataLayer;
+
+ public abstract DataLayer getDataLayer(CaseInsensitiveStringMap options);
+
+ DataLayer getDataLayerInternal(CaseInsensitiveStringMap options)
+ {
+ DataLayer dataLayer = this.dataLayer;
+ if (dataLayer != null)
+ {
+ return dataLayer;
+ }
+ dataLayer = getDataLayer(options);
+ this.dataLayer = dataLayer;
+ return dataLayer;
+ }
+
+ @Override
+ public StructType inferSchema(CaseInsensitiveStringMap options)
+ {
+ return getDataLayerInternal(options).structType();
+ }
+
+ @Override
+ public Table getTable(StructType schema, Transform[] partitioning, Map<String, String> properties)
+ {
+ return new CassandraTable(getDataLayerInternal(new CaseInsensitiveStringMap(properties)), schema);
+ }
+}
diff --git a/cassandra-analytics-sidecar-client/build.gradle b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/LocalDataSource.java
similarity index 56%
copy from cassandra-analytics-sidecar-client/build.gradle
copy to cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/LocalDataSource.java
index bedf2c04..c9d14e86 100644
--- a/cassandra-analytics-sidecar-client/build.gradle
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/LocalDataSource.java
@@ -1,5 +1,3 @@
-import java.nio.file.Paths
-
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -19,26 +17,27 @@ import java.nio.file.Paths
* under the License.
*/
-plugins {
- id('java-library')
- id('maven-publish')
-}
+package org.apache.cassandra.spark.sparksql;
-configurations {
- all*.exclude(group: 'org.slf4j', module: 'slf4j-log4j12')
- all*.exclude(group: 'log4j', module: 'log4j')
-}
+import org.apache.cassandra.spark.data.DataLayer;
+import org.apache.cassandra.spark.data.LocalDataLayer;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+import org.jetbrains.annotations.NotNull;
-if (propertyWithDefault("artifactType", null) == "common")
+@SuppressWarnings("unused")
+public class LocalDataSource extends CassandraTableProvider
{
- apply from: "$rootDir/gradle/common/publishing.gradle"
-}
+ @Override
+ @NotNull
+ public String shortName()
+ {
+ return "localsstabledatasource";
+ }
-dependencies {
- compileOnly("com.google.guava:guava:${guavaVersion}")
- implementation project(":cassandra-analytics-common")
-
- implementation "org.slf4j:slf4j-api:${slf4jApiVersion}"
-
- api(project(path: ':analytics-sidecar-vertx-client-shaded', configuration: 'shadow'))
-}
\ No newline at end of file
+ @Override
+ @NotNull
+ public DataLayer getDataLayer(@NotNull CaseInsensitiveStringMap options)
+ {
+ return LocalDataLayer.from(options);
+ }
+}
diff --git a/cassandra-analytics-sidecar-client/build.gradle b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/LocalPartitionSizeSource.java
similarity index 56%
copy from cassandra-analytics-sidecar-client/build.gradle
copy to cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/LocalPartitionSizeSource.java
index bedf2c04..cf24f1ce 100644
--- a/cassandra-analytics-sidecar-client/build.gradle
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/LocalPartitionSizeSource.java
@@ -1,5 +1,3 @@
-import java.nio.file.Paths
-
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -19,26 +17,23 @@ import java.nio.file.Paths
* under the License.
*/
-plugins {
- id('java-library')
- id('maven-publish')
-}
+package org.apache.cassandra.spark.sparksql;
-configurations {
- all*.exclude(group: 'org.slf4j', module: 'slf4j-log4j12')
- all*.exclude(group: 'log4j', module: 'log4j')
-}
+import org.apache.cassandra.spark.data.DataLayer;
+import org.apache.cassandra.spark.data.LocalDataLayer;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
-if (propertyWithDefault("artifactType", null) == "common")
+public class LocalPartitionSizeSource extends PartitionSizeTableProvider
{
- apply from: "$rootDir/gradle/common/publishing.gradle"
-}
+ @Override
+ public String shortName()
+ {
+ return "localpartitionsizesource";
+ }
-dependencies {
- compileOnly("com.google.guava:guava:${guavaVersion}")
- implementation project(":cassandra-analytics-common")
-
- implementation "org.slf4j:slf4j-api:${slf4jApiVersion}"
-
- api(project(path: ':analytics-sidecar-vertx-client-shaded', configuration: 'shadow'))
-}
\ No newline at end of file
+ @Override
+ public DataLayer getDataLayer(CaseInsensitiveStringMap options)
+ {
+ return LocalDataLayer.from(options);
+ }
+}
diff --git a/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/PartitionSizeIterator.java b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/PartitionSizeIterator.java
new file mode 100644
index 00000000..d78ed5fb
--- /dev/null
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/PartitionSizeIterator.java
@@ -0,0 +1,96 @@
+/*
+ * 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.cassandra.spark.sparksql;
+
+import java.io.IOException;
+
+import org.apache.cassandra.spark.data.CqlTable;
+import org.apache.cassandra.spark.data.DataLayer;
+import org.apache.cassandra.spark.data.converter.SparkSqlTypeConverter;
+import org.apache.cassandra.spark.reader.IndexEntry;
+import org.apache.cassandra.spark.reader.StreamScanner;
+import org.apache.cassandra.analytics.stats.Stats;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.connector.read.PartitionReader;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Wrapper iterator around IndexIterator to read all Index.db files and return SparkSQL
+ * rows containing all partition keys and the associated on-disk uncompressed and compressed sizes.
+ */
+public class PartitionSizeIterator implements PartitionReader<InternalRow>
+{
+ private final StreamScanner<IndexEntry> it;
+ private final CqlTable cqlTable;
+ private final int numPartitionKeys;
+ private final Stats stats;
+ private final long startTimeNanos;
+ private GenericInternalRow curr = null;
+ private final SparkSqlTypeConverter sparkSqlTypeConverter;
+
+ public PartitionSizeIterator(int partitionId, @NotNull DataLayer dataLayer)
+ {
+ this.cqlTable = dataLayer.cqlTable();
+ this.numPartitionKeys = cqlTable.numPartitionKeys();
+ this.stats = dataLayer.stats();
+ this.startTimeNanos = System.nanoTime();
+ this.it = dataLayer.openPartitionSizeIterator(partitionId);
+ stats.openedPartitionSizeIterator(System.nanoTime() - startTimeNanos);
+ this.sparkSqlTypeConverter = dataLayer.typeConverter();
+ }
+
+ /**
+ * The expected schema is defined in {@link DataLayer#partitionSizeStructType()}.
+ * It consists of the Cassandra partition keys, appended with the columns "uncompressed" and "compressed".
+ */
+ public boolean next() throws IOException
+ {
+ if (it.next())
+ {
+ it.advanceToNextColumn();
+
+ IndexEntry entry = it.data();
+ Object[] values = new Object[numPartitionKeys + 2];
+
+ CellIterator.readPartitionKey(sparkSqlTypeConverter, entry.getPartitionKey(), cqlTable, values, stats);
+ values[numPartitionKeys] = entry.getUncompressed();
+ values[numPartitionKeys + 1] = entry.getCompressed();
+
+ this.curr = new GenericInternalRow(values);
+ stats.emitIndexEntry(entry);
+
+ return true;
+ }
+
+ return false;
+ }
+
+ public InternalRow get()
+ {
+ return curr;
+ }
+
+ public void close() throws IOException
+ {
+ this.it.close();
+ stats.closedPartitionSizeIterator(System.nanoTime() - startTimeNanos);
+ }
+}
diff --git a/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/PartitionSizeTableProvider.java b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/PartitionSizeTableProvider.java
new file mode 100644
index 00000000..2acc567f
--- /dev/null
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/PartitionSizeTableProvider.java
@@ -0,0 +1,171 @@
+/*
+ * 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.cassandra.spark.sparksql;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import com.google.common.collect.ImmutableSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.spark.data.DataLayer;
+import org.apache.spark.TaskContext;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.SupportsRead;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableCapability;
+import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.connector.read.Batch;
+import org.apache.spark.sql.connector.read.InputPartition;
+import org.apache.spark.sql.connector.read.PartitionReader;
+import org.apache.spark.sql.connector.read.PartitionReaderFactory;
+import org.apache.spark.sql.connector.read.Scan;
+import org.apache.spark.sql.connector.read.ScanBuilder;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+public abstract class PartitionSizeTableProvider extends CassandraTableProvider
+{
+ public abstract DataLayer getDataLayer(CaseInsensitiveStringMap options);
+
+ @Override
+ public StructType inferSchema(CaseInsensitiveStringMap options)
+ {
+ return getDataLayerInternal(options).partitionSizeStructType();
+ }
+
+ @Override
+ public Table getTable(StructType schema, Transform[] partitioning, Map<String, String> properties)
+ {
+ return new PartitionSizeTable(getDataLayerInternal(new CaseInsensitiveStringMap(properties)), schema);
+ }
+}
+
+class PartitionSizeTable implements Table, SupportsRead
+{
+ private final DataLayer dataLayer;
+ private final StructType schema;
+
+ PartitionSizeTable(DataLayer dataLayer, StructType schema)
+ {
+ this.dataLayer = dataLayer;
+ this.schema = schema;
+ }
+
+ @Override
+ public String name()
+ {
+ return dataLayer.cqlTable().keyspace() + "." + dataLayer.cqlTable().table();
+ }
+
+ @Override
+ public StructType schema()
+ {
+ return schema;
+ }
+
+ @Override
+ public Set<TableCapability> capabilities()
+ {
+ return ImmutableSet.of(TableCapability.BATCH_READ);
+ }
+
+ @Override
+ public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options)
+ {
+ return new PartitionSizeScanBuilder(dataLayer, schema, options);
+ }
+}
+
+class PartitionSizeScanBuilder implements ScanBuilder, Scan, Batch
+{
+ final DataLayer dataLayer;
+ final StructType schema;
+ final CaseInsensitiveStringMap options;
+
+ PartitionSizeScanBuilder(DataLayer dataLayer, StructType schema, CaseInsensitiveStringMap options)
+ {
+ this.dataLayer = dataLayer;
+ this.schema = schema;
+ this.options = options;
+ }
+
+ @Override
+ public Scan build()
+ {
+ return this;
+ }
+
+ @Override
+ public StructType readSchema()
+ {
+ return schema;
+ }
+
+ @Override
+ public Batch toBatch()
+ {
+ return this;
+ }
+
+ @Override
+ public InputPartition[] planInputPartitions()
+ {
+ return IntStream.range(0, this.dataLayer.partitionCount())
+ .mapToObj(CassandraInputPartition::new)
+ .toArray(InputPartition[]::new);
+ }
+
+ @Override
+ public PartitionReaderFactory createReaderFactory()
+ {
+ return new PartitionSizeReaderFactory(dataLayer);
+ }
+}
+
+class PartitionSizeReaderFactory implements PartitionReaderFactory
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(CassandraPartitionReaderFactory.class);
+ final DataLayer dataLayer;
+
+ PartitionSizeReaderFactory(DataLayer dataLayer)
+ {
+ this.dataLayer = dataLayer;
+ }
+
+ @Override
+ public PartitionReader<InternalRow> createReader(InputPartition partition)
+ {
+ int partitionId;
+ if (partition instanceof CassandraInputPartition)
+ {
+ partitionId = ((CassandraInputPartition) partition).getPartitionId();
+ }
+ else
+ {
+ partitionId = TaskContext.getPartitionId();
+ LOGGER.warn("InputPartition is not of CassandraInputPartition type. Using TaskContext to determine the partitionId type={}, partitionId={}",
+ partition.getClass().getName(), partitionId);
+ }
+ return new PartitionSizeIterator(partitionId, dataLayer);
+ }
+}
diff --git a/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/SparkRowIterator.java b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/SparkRowIterator.java
new file mode 100644
index 00000000..d076b342
--- /dev/null
+++ b/cassandra-analytics-core/src/main/spark4/org/apache/cassandra/spark/sparksql/SparkRowIterator.java
@@ -0,0 +1,87 @@
+/*
+ * 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.cassandra.spark.sparksql;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.cassandra.spark.config.SchemaFeature;
+import org.apache.cassandra.spark.data.DataLayer;
+import org.apache.cassandra.spark.sparksql.filters.PartitionKeyFilter;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.connector.read.PartitionReader;
+import org.apache.spark.sql.types.StructType;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Wrapper iterator around SparkCellIterator to normalize cells into Spark SQL rows
+ */
+public class SparkRowIterator extends AbstractSparkRowIterator<GenericInternalRow> implements PartitionReader<InternalRow>
+{
+ @VisibleForTesting
+ public SparkRowIterator(int partitionId, @NotNull DataLayer dataLayer)
+ {
+ this(partitionId, dataLayer, null, new ArrayList<>());
+ }
+
+ public SparkRowIterator(int partitionId,
+ @NotNull DataLayer dataLayer,
+ @Nullable StructType requiredSchema,
+ @NotNull List<PartitionKeyFilter> partitionKeyFilters)
+ {
+ super(
+ partitionId,
+ dataLayer,
+ requiredSchema,
+ partitionKeyFilters,
+ (builder) -> decorate(requiredSchema, builder, dataLayer.requestedFeatures())
+ );
+ }
+
+ protected static RowBuilder<GenericInternalRow> decorate(@Nullable StructType requiredSchema,
+ RowBuilder<GenericInternalRow> builder,
+ List<SchemaFeature> features)
+ {
+ Set<String> fieldNames = requiredSchema == null ? null : new HashSet<>(Arrays.asList(requiredSchema.fieldNames()));
+ for (SchemaFeature feature : features)
+ {
+ // Only decorate when there is no column filter or when the field is requested in the query,
+ // otherwise we skip decoration
+ if (fieldNames == null || fieldNames.contains(feature.fieldName()))
+ {
+ builder = feature.decorate(builder);
+ }
+ }
+
+ return builder;
+ }
+
+ public GenericInternalRow rowBuilder(Object[] valueArray)
+ {
+ return new GenericInternalRow(valueArray);
+ }
+}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/SSTableReaderTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/SSTableReaderTests.java
index 53072e17..96248ad1 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/SSTableReaderTests.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/SSTableReaderTests.java
@@ -29,17 +29,16 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import com.google.common.util.concurrent.Uninterruptibles;
-import org.apache.commons.lang.StringUtils;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
+import org.apache.cassandra.analytics.stats.Stats;
import org.apache.cassandra.bridge.CassandraBridge;
import org.apache.cassandra.spark.data.BasicSupplier;
import org.apache.cassandra.spark.data.CqlTable;
import org.apache.cassandra.spark.data.FileType;
import org.apache.cassandra.spark.reader.RowData;
import org.apache.cassandra.spark.reader.StreamScanner;
-import org.apache.cassandra.analytics.stats.Stats;
import org.apache.cassandra.spark.sparksql.filters.SSTableTimeRangeFilter;
import org.apache.cassandra.spark.utils.ByteBufferUtils;
import org.apache.cassandra.spark.utils.TimeProvider;
@@ -123,7 +122,7 @@ public class SSTableReaderTests
ByteBuffer colBuf = rowData.getColumnName();
String colName = ByteBufferUtils.string(ByteBufferUtils.readBytesWithShortLength(colBuf));
colBuf.get();
- if (StringUtils.isEmpty(colName))
+ if (colName == null || colName.isEmpty())
{
continue;
}
@@ -230,7 +229,7 @@ public class SSTableReaderTests
ByteBuffer colBuf = rowData.getColumnName();
String colName = ByteBufferUtils.string(ByteBufferUtils.readBytesWithShortLength(colBuf));
colBuf.get();
- if (StringUtils.isEmpty(colName))
+ if (colName == null || colName.isEmpty())
{
continue;
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java
index ab0e663b..5271e00e 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java
@@ -127,7 +127,12 @@ class BulkSparkConfTest
BulkSparkConf.setupSparkConf(sparkConf, true);
assertThat(sparkConf.get("spark.kryo.registrator", ""))
.isEqualTo("," + SbwKryoRegistrator.class.getName());
- if (BuildInfo.isAtLeastJava11(BuildInfo.javaSpecificationVersion()))
+ if (BuildInfo.isAtLeastJava17(BuildInfo.javaSpecificationVersion()))
+ {
+ assertThat(sparkConf.get("spark.executor.extraJavaOptions", ""))
+ .isEqualTo(BulkSparkConf.JDK11_OPTIONS + BulkSparkConf.JDK17_OPTIONS);
+ }
+ else if (BuildInfo.isAtLeastJava11(BuildInfo.javaSpecificationVersion()))
{
assertThat(sparkConf.get("spark.executor.extraJavaOptions", ""))
.isEqualTo(BulkSparkConf.JDK11_OPTIONS);
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java
index 1c2d6018..f6fe8c5e 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java
@@ -124,7 +124,7 @@ public class DirectStreamSessionTest
replicaAwareFailureHandler(), null)
)
.isInstanceOf(IllegalStateException.class)
- .hasMessage("No replicas found for range (0‥0]");
+ .hasMessageMatching("No replicas found for range \\(0(‥|..)0]");
}
@Test
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java
index 8a2c1728..b1ce66e3 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java
@@ -394,10 +394,10 @@ class RecordWriterTest
(wc, path, dp, pid) -> new SortedSSTableWriter(tw, folder, digestAlgorithm, pid));
Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, Range.all(), false, false, false);
String expectedErr = "java.lang.IllegalStateException: Received Token " +
- "5765203080415074583 outside the expected ranges [(-9223372036854775808‥100000]]";
+ "5765203080415074583 outside the expected ranges \\[\\(-9223372036854775808(‥|..)100000]]";
assertThatThrownBy(() -> rw.write(data))
.isInstanceOf(RuntimeException.class)
- .hasMessage(expectedErr);
+ .hasMessageMatching(expectedErr);
}
@Test
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriterTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriterTest.java
index 4bb3f7dc..271c760d 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriterTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriterTest.java
@@ -36,7 +36,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
-import javax.validation.constraints.NotNull;
import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.Uninterruptibles;
@@ -53,6 +52,7 @@ import org.apache.cassandra.spark.bulkwriter.token.ConsistencyLevel;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.cassandra.spark.common.Digest;
import org.apache.cassandra.spark.utils.XXHash32DigestAlgorithm;
+import org.jetbrains.annotations.NotNull;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java
index e907f105..64ee81b7 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java
@@ -26,7 +26,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
-import javax.validation.constraints.NotNull;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableMap;
@@ -42,6 +41,7 @@ import org.apache.cassandra.spark.common.model.NodeStatus;
import org.apache.cassandra.spark.data.ReplicationFactor;
import org.apache.cassandra.spark.data.partitioner.Partitioner;
import org.apache.cassandra.spark.utils.RangeUtils;
+import org.jetbrains.annotations.NotNull;
public final class TokenRangeMappingUtils
{
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/ClientConfigTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/ClientConfigTests.java
index b42e33d2..2d6a5896 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/ClientConfigTests.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/ClientConfigTests.java
@@ -206,6 +206,6 @@ class ClientConfigTests
assertThatThrownBy(() -> ClientConfig.create(options))
.isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Invalid range: [2000‥1000]");
+ .hasMessageMatching("Invalid range: \\[2000(‥|..)1000]");
}
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/endtoend/SchemaTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/endtoend/SchemaTests.java
index 08ab169f..f2d452ad 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/endtoend/SchemaTests.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/endtoend/SchemaTests.java
@@ -19,11 +19,12 @@
package org.apache.cassandra.spark.endtoend;
+import java.math.BigInteger;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -43,6 +44,8 @@ import org.apache.cassandra.spark.data.CqlField;
import org.apache.cassandra.spark.utils.RandomUtils;
import org.apache.cassandra.spark.utils.test.TestSchema;
import org.apache.spark.sql.Row;
+import org.apache.spark.sql.functions;
+import org.apache.spark.sql.types.DataTypes;
import static org.assertj.core.api.Assertions.assertThat;
import static org.quicktheories.QuickTheory.qt;
@@ -284,7 +287,11 @@ public class SchemaTests
.withClusteringKey("c", bridge.text())
.withColumn("d", bridge.text())
.withColumn("e", bridge.bigint());
- AtomicLong total = new AtomicLong(0);
+ // AtomicLong is not sufficient here, using BigInteger instead. The aggregation
+ // in the withCheck block will overflow and this will throw an exception in Spark 4.
+ // In earlier versions of Spark the long value would just overflow and the aggregated
+ // value was incorrect.
+ AtomicReference<BigInteger> total = new AtomicReference<>(BigInteger.ZERO);
Map<UUID, TestSchema.TestRow> rows = new HashMap<>(Tester.DEFAULT_NUM_ROWS);
Tester.builder(schemaBuilder)
// Don't write random data
@@ -319,7 +326,7 @@ public class SchemaTests
TestSchema.TestRow newTestRow = testRow.copy("e", RandomUtils.RANDOM.nextLong())
.copy("d", UUID.randomUUID().toString().substring(0, 10));
rows.put(testRow.getUUID("a"), newTestRow);
- total.addAndGet(newTestRow.getLong("e"));
+ total.updateAndGet(t -> t.add(BigInteger.valueOf(newTestRow.getLong("e"))));
writer.write(newTestRow.allValues());
}
})
@@ -328,10 +335,17 @@ public class SchemaTests
.withReadListener(actualRow -> assertThat(actualRow).isEqualTo(rows.get(actualRow.getUUID("a"))))
.withReadListener(actualRow -> assertThat(actualRow.getLong("e")).isEqualTo(rows.get(actualRow.getUUID("a")).getLong("e")))
// Verify Spark aggregations match expected
- .withCheck(dataset -> assertThat(dataset.groupBy().sum("e").first().getLong(0)).isEqualTo(total.get()))
+ .withCheck(dataset -> {
+ BigInteger sum = dataset.agg(functions.sum(
+ functions.col("e").cast(DataTypes.createDecimalType(38, 0))))
+ .first()
+ .getDecimal(0)
+ .toBigInteger();
+ assertThat(sum).isEqualTo(total.get());
+ })
.withCheck(dataset -> assertThat(dataset.groupBy().count().first().getLong(0)).isEqualTo(rows.size()))
.withReset(() -> {
- total.set(0);
+ total.set(BigInteger.ZERO);
rows.clear();
})
.run(bridge.getVersion());
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BufferingInputStreamTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BufferingInputStreamTests.java
index fc41d7fb..b7c30524 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BufferingInputStreamTests.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BufferingInputStreamTests.java
@@ -36,13 +36,11 @@ import java.util.stream.IntStream;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.common.util.concurrent.Uninterruptibles;
-import org.apache.commons.lang.ArrayUtils;
-import org.apache.commons.lang.mutable.MutableInt;
import org.junit.jupiter.api.Test;
+import org.apache.cassandra.analytics.stats.Stats;
import org.apache.cassandra.spark.data.FileType;
import org.apache.cassandra.spark.data.SSTable;
-import org.apache.cassandra.analytics.stats.Stats;
import org.apache.cassandra.spark.utils.streaming.BufferingInputStream;
import org.apache.cassandra.spark.utils.streaming.CassandraFileSource;
import org.apache.cassandra.spark.utils.streaming.StreamBuffer;
@@ -61,9 +59,9 @@ public class BufferingInputStreamTests
{
private static final ScheduledExecutorService SCHEDULER = Executors.newScheduledThreadPool(1);
private static final ExecutorService EXECUTOR =
- Executors.newFixedThreadPool(4, new ThreadFactoryBuilder().setNameFormat("sstable-tests-%d")
- .setDaemon(true)
- .build());
+ Executors.newFixedThreadPool(4, new ThreadFactoryBuilder().setNameFormat("sstable-tests-%d")
+ .setDaemon(true)
+ .build());
static final int DEFAULT_CHUNK_SIZE = 8192;
static final Stats STATS = Stats.DoNothingStats.INSTANCE;
@@ -139,9 +137,9 @@ public class BufferingInputStreamTests
maxBufferSize,
requestChunkSize,
(start, end, consumer) -> {
- requestCount.incrementAndGet();
- writeBuffers(consumer, randomBuffers(chunksPerRequest));
- }, null);
+ requestCount.incrementAndGet();
+ writeBuffers(consumer, randomBuffers(chunksPerRequest));
+ }, null);
BufferingInputStream<SSTable> is = new BufferingInputStream<>(mockedClient, STATS.bufferingInputStreamStats());
readStreamFully(is);
assertThat(requestCount.get()).isEqualTo(numRequests);
@@ -161,18 +159,18 @@ public class BufferingInputStreamTests
CassandraFileSource.DEFAULT_MAX_BUFFER_SIZE,
CassandraFileSource.DEFAULT_CHUNK_BUFFER_SIZE,
(start, end, consumer) -> {
- if (count.incrementAndGet() > (numRequests / 2))
- {
- // Halfway through throw random exception
- EXECUTOR.submit(() -> consumer.onError(new RuntimeException("Something bad happened...")));
- }
- else
- {
- writeBuffers(consumer, randomBuffers(chunksPerRequest));
- }
- }, null);
+ if (count.incrementAndGet() > (numRequests / 2))
+ {
+ // Halfway through throw random exception
+ EXECUTOR.submit(() -> consumer.onError(new RuntimeException("Something bad happened...")));
+ }
+ else
+ {
+ writeBuffers(consumer, randomBuffers(chunksPerRequest));
+ }
+ }, null);
assertThatThrownBy(() -> readStreamFully(new BufferingInputStream<>(source, STATS.bufferingInputStreamStats())))
- .isInstanceOf(IOException.class);
+ .isInstanceOf(IOException.class);
}
@Test
@@ -180,21 +178,21 @@ public class BufferingInputStreamTests
{
long now = System.nanoTime();
assertThat(timeoutLeftNanos(Duration.ofMillis(1000), now, now - Duration.ofMillis(900).toNanos()))
- .isEqualTo(Duration.ofMillis(100).toNanos());
+ .isEqualTo(Duration.ofMillis(100).toNanos());
assertThat(timeoutLeftNanos(Duration.ofMillis(1000), now, now - Duration.ofMillis(1500).toNanos()))
- .isEqualTo(Duration.ofMillis(-500).toNanos());
+ .isEqualTo(Duration.ofMillis(-500).toNanos());
assertThat(timeoutLeftNanos(Duration.ofMillis(1000), now, now - Duration.ofMillis(5).toNanos()))
- .isEqualTo(Duration.ofMillis(995).toNanos());
+ .isEqualTo(Duration.ofMillis(995).toNanos());
assertThat(timeoutLeftNanos(Duration.ofMillis(1000), now, now - Duration.ofMillis(0).toNanos()))
- .isEqualTo(Duration.ofMillis(1000).toNanos());
+ .isEqualTo(Duration.ofMillis(1000).toNanos());
assertThat(timeoutLeftNanos(Duration.ofMillis(1000), now, now + Duration.ofMillis(500).toNanos()))
- .isEqualTo(Duration.ofMillis(1000).toNanos());
+ .isEqualTo(Duration.ofMillis(1000).toNanos());
assertThat(timeoutLeftNanos(Duration.ofMillis(60000), now, now - Duration.ofMillis(25000).toNanos()))
- .isEqualTo(Duration.ofMillis(35000).toNanos());
+ .isEqualTo(Duration.ofMillis(35000).toNanos());
assertThat(timeoutLeftNanos(Duration.ofMillis(60000), now, now - Duration.ofMillis(65000).toNanos()))
- .isEqualTo(Duration.ofMillis(-5000).toNanos());
+ .isEqualTo(Duration.ofMillis(-5000).toNanos());
assertThat(timeoutLeftNanos(Duration.ofMillis(60000), now, now - Duration.ofMillis(60000).toNanos()))
- .isEqualTo(Duration.ofMillis(0).toNanos());
+ .isEqualTo(Duration.ofMillis(0).toNanos());
}
@Test
@@ -212,15 +210,15 @@ public class BufferingInputStreamTests
CassandraFileSource.DEFAULT_MAX_BUFFER_SIZE,
CassandraFileSource.DEFAULT_CHUNK_BUFFER_SIZE,
(start, end, consumer) -> {
- // Only respond once so future requests will time out
- if (count.incrementAndGet() == 1)
- {
- EXECUTOR.submit(() -> {
- Uninterruptibles.sleepUninterruptibly(sleepTimeInMillis, TimeUnit.MILLISECONDS);
- writeBuffers(consumer, randomBuffers(chunksPerRequest));
- });
- }
- }, timeout);
+ // Only respond once so future requests will time out
+ if (count.incrementAndGet() == 1)
+ {
+ EXECUTOR.submit(() -> {
+ Uninterruptibles.sleepUninterruptibly(sleepTimeInMillis, TimeUnit.MILLISECONDS);
+ writeBuffers(consumer, randomBuffers(chunksPerRequest));
+ });
+ }
+ }, timeout);
BufferingInputStream<SSTable> inputStream = new BufferingInputStream<>(source, STATS.bufferingInputStreamStats());
try
{
@@ -234,9 +232,9 @@ public class BufferingInputStreamTests
long readAndTimeoutTotal = TimeUnit.NANOSECONDS.toMillis(inputStream.timeBlockedNanos()) + timeout.toMillis();
Duration clientTimeoutTotal = Duration.ofNanos(System.nanoTime() - startTime);
assertThat(clientTimeoutTotal.toMillis()).isGreaterThanOrEqualTo(readAndTimeoutTotal)
- .describedAs("Timeout didn't account for activity time. "
- + "Took %dms should have taken at least %dms",
- clientTimeoutTotal.toMillis(), readAndTimeoutTotal);
+ .describedAs("Timeout didn't account for activity time. "
+ + "Took %dms should have taken at least %dms",
+ clientTimeoutTotal.toMillis(), readAndTimeoutTotal);
}
@Test
@@ -245,8 +243,8 @@ public class BufferingInputStreamTests
int size = 20971520;
int chunkSize = 1024;
int numChunks = 16;
- MutableInt bytesRead = new MutableInt(0);
- MutableInt count = new MutableInt(0);
+ AtomicInteger bytesRead = new AtomicInteger(0);
+ AtomicInteger count = new AtomicInteger(0);
CassandraFileSource<SSTable> source = new CassandraFileSource<SSTable>()
{
@Override
@@ -255,8 +253,8 @@ public class BufferingInputStreamTests
assertThat(start).isNotEqualTo(0);
int length = (int) (end - start + 1);
consumer.onRead(randomBuffer(length));
- bytesRead.add(length);
- count.increment();
+ bytesRead.addAndGet(length);
+ count.incrementAndGet();
consumer.onEnd();
}
@@ -416,7 +414,7 @@ public class BufferingInputStreamTests
buffer.limit(remainingReadBytes);
read = stream2.read(buffer); // read remaining bytes
assertThat(read).isEqualTo(remainingReadBytes);
- assertThat(ArrayUtils.subarray(buffer.array(), 0, remainingReadBytes)).isEqualTo(returnedBuffers.get(1));
+ assertThat(buffer.array()).startsWith(returnedBuffers.get(1));
}
}
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BuildInfoTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BuildInfoTest.java
index 10681e73..ab860527 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BuildInfoTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BuildInfoTest.java
@@ -20,6 +20,8 @@
package org.apache.cassandra.spark.utils;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import static org.assertj.core.api.Assertions.assertThat;
@@ -49,30 +51,38 @@ public class BuildInfoTest
}
@Test
- public void isAtLeastJava11()
+ public void testIsAtLeastJavaVersionWithNullInput()
{
assertThat(BuildInfo.isAtLeastJava11(null)).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("0.9")).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("1.1")).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("1.2")).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("1.3")).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("1.4")).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("1.5")).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("1.6")).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("1.7")).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("1.8")).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("9")).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("10")).isFalse();
- assertThat(BuildInfo.isAtLeastJava11("11")).isTrue();
- assertThat(BuildInfo.isAtLeastJava11("12")).isTrue();
- assertThat(BuildInfo.isAtLeastJava11("13")).isTrue();
- assertThat(BuildInfo.isAtLeastJava11("14")).isTrue();
- assertThat(BuildInfo.isAtLeastJava11("15")).isTrue();
- assertThat(BuildInfo.isAtLeastJava11("16")).isTrue();
- assertThat(BuildInfo.isAtLeastJava11("17")).isTrue();
- assertThat(BuildInfo.isAtLeastJava11("18")).isTrue();
- assertThat(BuildInfo.isAtLeastJava11("19")).isTrue();
- assertThat(BuildInfo.isAtLeastJava11("20")).isTrue();
- assertThat(BuildInfo.isAtLeastJava11("21")).isTrue();
+ assertThat(BuildInfo.isAtLeastJava17(null)).isFalse();
+ }
+
+ @ParameterizedTest(name = "{index} => Java version {0}")
+ @ValueSource(strings = { "0.9", "1.1", "1.2", "1.3", "1.4", "1.5", "1.6", "1.7", "1.8", "9", "10" })
+ public void isNotAtLeastJava11(String version)
+ {
+ assertThat(BuildInfo.isAtLeastJava11(version)).isFalse();
+ }
+
+ @ParameterizedTest(name = "{index} => Java version {0}")
+ @ValueSource(strings = { "11", "12", "13", "14", "15", "16", "17", "18", "19", "20", "21", "22", "23", "24", "25" })
+ public void isAtLeastJava11(String version)
+ {
+ assertThat(BuildInfo.isAtLeastJava11(version)).isTrue();
+ }
+
+ @ParameterizedTest(name = "{index} => Java version {0}")
+ @ValueSource(strings = { "0.9", "1.1", "1.2", "1.3", "1.4", "1.5", "1.6", "1.7", "1.8", "9", "10",
+ "11", "12", "13", "14", "15", "16" })
+ public void isNotAtLeastJava17(String version)
+ {
+ assertThat(BuildInfo.isAtLeastJava17(version)).isFalse();
+ }
+
+ @ParameterizedTest(name = "{index} => Java version {0}")
+ @ValueSource(strings = { "17", "18", "19", "20", "21", "22", "23", "24", "25" })
+ public void isAtLeastJava17(String version)
+ {
+ assertThat(BuildInfo.isAtLeastJava17(version)).isTrue();
}
}
diff --git a/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/PartitionSizeTests.java b/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/PartitionSizeTests.java
new file mode 100644
index 00000000..d3a702ef
--- /dev/null
+++ b/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/PartitionSizeTests.java
@@ -0,0 +1,95 @@
+/*
+ * 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.cassandra.spark;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import org.apache.cassandra.bridge.CassandraVersion;
+import org.apache.cassandra.spark.data.VersionRunner;
+import org.apache.cassandra.spark.utils.test.TestSchema;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class PartitionSizeTests extends VersionRunner
+{
+ @ParameterizedTest
+ @MethodSource("org.apache.cassandra.spark.data.VersionRunner#versions")
+ public void testReadingPartitionSize(CassandraVersion version)
+ {
+ TestUtils.runTest(version, (partitioner, dir, bridge) -> {
+ int numRows = Tester.DEFAULT_NUM_ROWS;
+ int numCols = 25;
+ TestSchema schema = TestSchema.builder(bridge)
+ .withPartitionKey("a", bridge.text())
+ .withClusteringKey("b", bridge.aInt())
+ .withColumn("c", bridge.aInt())
+ .withColumn("d", bridge.text()).build();
+
+ Map<String, Integer> sizes = new HashMap<>(numRows);
+ schema.writeSSTable(dir, bridge, partitioner, (writer) -> {
+ for (int i = 0; i < numRows; i++)
+ {
+ String key = UUID.randomUUID().toString();
+ int size = 0;
+ for (int j = 0; j < numCols; j++)
+ {
+ String str = TestUtils.randomLowEntropyString();
+ writer.write(key, j, i + j, str);
+ size += 4 + 4 + str.getBytes(StandardCharsets.UTF_8).length;
+ }
+ sizes.put(key, size);
+ }
+ });
+
+ Dataset<Row> ds = TestUtils.openLocalPartitionSizeSource(bridge,
+ partitioner,
+ dir,
+ schema.keyspace,
+ schema.createStatement,
+ version,
+ Collections.emptySet(),
+ null);
+ List<Row> rows = ds.collectAsList();
+ assertThat(rows).hasSize(numRows);
+ for (Row row : rows)
+ {
+ String key = row.getString(0);
+ long uncompressed = row.getLong(1);
+ long compressed = row.getLong(2);
+ assertThat(sizes).containsKey(key);
+ long len = sizes.get(key);
+ assertThat(len).isLessThan(uncompressed);
+ assertThat(Math.abs(uncompressed - len)).isLessThan(500); // uncompressed size should be ~len size but with a fixed overhead
+ assertThat(compressed).isLessThan(uncompressed);
+ assertThat(compressed / (float) uncompressed).isLessThan(0.1f);
+ }
+ });
+ }
+}
diff --git a/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/bulkwriter/TestTaskContext.java b/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/bulkwriter/TestTaskContext.java
new file mode 100644
index 00000000..7106bb95
--- /dev/null
+++ b/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/bulkwriter/TestTaskContext.java
@@ -0,0 +1,218 @@
+/*
+ * 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.cassandra.spark.bulkwriter;
+
+import java.io.Closeable;
+import java.util.Properties;
+
+import org.apache.spark.TaskContext;
+import org.apache.spark.executor.TaskMetrics;
+import org.apache.spark.memory.TaskMemoryManager;
+import org.apache.spark.metrics.source.Source;
+import org.apache.spark.resource.ResourceInformation;
+import org.apache.spark.shuffle.FetchFailedException;
+import org.apache.spark.util.AccumulatorV2;
+import org.apache.spark.util.TaskCompletionListener;
+import org.apache.spark.util.TaskFailureListener;
+import scala.Function0;
+import scala.Option;
+import scala.collection.immutable.Map;
+import scala.collection.immutable.Seq;
+
+public class TestTaskContext extends TaskContext
+{
+ @Override
+ public boolean isCompleted()
+ {
+ return false;
+ }
+
+ @Override
+ public boolean isFailed()
+ {
+ return false;
+ }
+
+ @Override
+ public boolean isInterrupted()
+ {
+ return false;
+ }
+
+ @Deprecated
+ public boolean isRunningLocally()
+ {
+ return false;
+ }
+
+ @Override
+ public TaskContext addTaskCompletionListener(TaskCompletionListener listener)
+ {
+ return null;
+ }
+
+ @Override
+ public TaskContext addTaskFailureListener(TaskFailureListener listener)
+ {
+ return null;
+ }
+
+ @Override
+ public int stageId()
+ {
+ return 0;
+ }
+
+ @Override
+ public int stageAttemptNumber()
+ {
+ return 0;
+ }
+
+ @Override
+ public int partitionId()
+ {
+ return 0;
+ }
+
+ @Override
+ public int numPartitions()
+ {
+ return 0;
+ }
+
+ @Override
+ public int attemptNumber()
+ {
+ return 0;
+ }
+
+ @Override
+ public long taskAttemptId()
+ {
+ return 0;
+ }
+
+ @Override
+ public String getLocalProperty(String key)
+ {
+ return null;
+ }
+
+ @Override
+ public int cpus()
+ {
+ return 1;
+ }
+
+ public Map<String, ResourceInformation> resources()
+ {
+ return null;
+ }
+
+ public java.util.Map<String, ResourceInformation> resourcesJMap()
+ {
+ return null;
+ }
+
+ @Override
+ public TaskMetrics taskMetrics()
+ {
+ return null;
+ }
+
+ @Override
+ public Seq<Source> getMetricsSources(String sourceName)
+ {
+ return null;
+ }
+
+ @Override
+ public void killTaskIfInterrupted()
+ {
+ }
+
+ @Override
+ public Option<String> getKillReason()
+ {
+ return null;
+ }
+
+ @Override
+ public TaskMemoryManager taskMemoryManager()
+ {
+ return null;
+ }
+
+ @Override
+ public void registerAccumulator(AccumulatorV2 accumulator)
+ {
+ }
+
+ @Override
+ public void setFetchFailed(FetchFailedException fetchFailed)
+ {
+ }
+
+ @Override
+ public void markInterrupted(String reason)
+ {
+ }
+
+ @Override
+ public void markTaskFailed(Throwable error)
+ {
+ }
+
+ @Override
+ public void markTaskCompleted(Option<Throwable> error)
+ {
+ }
+
+ @Override
+ public Option<FetchFailedException> fetchFailed()
+ {
+ return null;
+ }
+
+ @Override
+ public Properties getLocalProperties()
+ {
+ return null;
+ }
+
+ @Override
+ public boolean interruptible()
+ {
+ return false;
+ }
+
+ @Override
+ public void pendingInterrupt(Option<Thread> threadToInterrupt, String reason)
+ {
+
+ }
+
+ @Override
+ public <T extends Closeable> T createResourceUninterruptibly(Function0<T> resourceBuilder)
+ {
+ return null;
+ }
+}
diff --git a/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/common/SidecarInstanceFactoryTest.java b/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/common/SidecarInstanceFactoryTest.java
new file mode 100644
index 00000000..277175a7
--- /dev/null
+++ b/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/common/SidecarInstanceFactoryTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.cassandra.spark.common;
+
+import org.junit.jupiter.api.Test;
+
+import o.a.c.sidecar.client.shaded.client.SidecarInstance;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class SidecarInstanceFactoryTest
+{
+ @Test
+ void testCreateSidecarInstance()
+ {
+ assertThatThrownBy(() -> SidecarInstanceFactory.createFromString("", 9999))
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Unable to create sidecar instance from empty input");
+
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("localhost", 9999),
+ "localhost", 9999);
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("[2024:a::1]", 9999),
+ "[2024:a::1]", 9999);
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("localhost:8888", 9999),
+ "localhost", 8888);
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("127.0.0.1:8888", 9999),
+ "127.0.0.1", 8888);
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("[2024:a::1]:8888", 9999),
+ "[2024:a::1]", 8888);
+ }
+
+ private void assertSidecarInstance(SidecarInstance sidecarInstance, String expectedHostname, int expectedPort)
+ {
+ assertThat(sidecarInstance.hostname()).isEqualTo(expectedHostname);
+ assertThat(sidecarInstance.port()).isEqualTo(expectedPort);
+ }
+}
diff --git a/cassandra-analytics-integration-tests/build.gradle b/cassandra-analytics-integration-tests/build.gradle
index 4604e8c5..0a01c84e 100644
--- a/cassandra-analytics-integration-tests/build.gradle
+++ b/cassandra-analytics-integration-tests/build.gradle
@@ -27,11 +27,12 @@ plugins {
id('maven-publish')
}
-if (propertyWithDefault("artifactType", null) == "spark")
-{
+if (propertyWithDefault("artifactType", null) == "spark") {
apply from: "$rootDir/gradle/common/publishing.gradle"
}
+apply from: "${project.rootDir}/gradle/common/javaOptions.gradle"
+
def integrationMaxHeapSize = System.getenv("INTEGRATION_MAX_HEAP_SIZE") ?: "3072M"
println("Using ${integrationMaxHeapSize} maxHeapSize")
@@ -125,26 +126,8 @@ def configureIntegrationTestTask = { Test task, String majorMinor = null ->
showStandardStreams = false
}
- if (JavaVersion.current().isJava11Compatible()) {
- def JDK11_OPTIONS = ['-Djdk.attach.allowAttachSelf=true',
- '--add-exports', 'java.base/jdk.internal.misc=ALL-UNNAMED',
- '--add-exports', 'java.base/jdk.internal.ref=ALL-UNNAMED',
- '--add-exports', 'java.base/sun.nio.ch=ALL-UNNAMED',
- '--add-exports', 'java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED',
- '--add-exports', 'java.rmi/sun.rmi.registry=ALL-UNNAMED',
- '--add-exports', 'java.rmi/sun.rmi.server=ALL-UNNAMED',
- '--add-exports', 'java.sql/java.sql=ALL-UNNAMED',
- '--add-opens', 'java.base/java.lang.module=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.loader=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.ref=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.reflect=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.math=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.module=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.util.jar=ALL-UNNAMED',
- '--add-opens', 'jdk.management/com.sun.management.internal=ALL-UNNAMED']
- task.jvmArgs(JDK11_OPTIONS)
- println("JVM arguments for $project.name are ${task.allJvmArgs}")
- }
+ task.jvmArgs(project.ext.JDK_OPTIONS)
+ println("JVM arguments for $project.name are ${task.allJvmArgs}")
// Some test classes are skipped entirely via assumeThat (e.g. MultipleTokens tests on C* 4.0).
// When CI runs individual classes with --tests, a skipped @BeforeAll leaves zero discovered tests,
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteDataTypesTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteDataTypesTest.java
index 34928792..49f43c30 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteDataTypesTest.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteDataTypesTest.java
@@ -85,11 +85,13 @@ class BulkWriteDataTypesTest extends SharedClusterSparkIntegrationTestBase
Dataset<Row> df = generateDataset(spark, typeTestSetup);
QualifiedName table = new QualifiedName(TEST_KEYSPACE, tableName);
- if (typeTestSetup.expectedFailureMessage != null)
+ if (typeTestSetup.expectedFailureMessages != null)
{
assertThatException()
.isThrownBy(() -> bulkWriterDataFrameWriter(df, table).save())
- .withMessageContaining(typeTestSetup.expectedFailureMessage);
+ .extracting(Throwable::getMessage)
+ .asString()
+ .containsAnyOf(typeTestSetup.expectedFailureMessages);
}
else
{
@@ -227,7 +229,10 @@ class BulkWriteDataTypesTest extends SharedClusterSparkIntegrationTestBase
Arrays.asList(IntegerType, CalendarIntervalType),
Arrays.asList(INTEGER_MAPPER, DURATION_MAPPER),
"CREATE TABLE %s (id int, took duration, PRIMARY KEY (id))",
- "Cannot save interval data type into external storage.");
+ "Cannot save interval data type into external storage",
+ // Error message changed in Cassandra 5.0
+ "datasource doesn't support the column `took` of the type \"INTERVAL\"."
+ );
}
static TypeTestSetup integersAndStringsSchemaSetup()
@@ -467,7 +472,7 @@ class BulkWriteDataTypesTest extends SharedClusterSparkIntegrationTestBase
final List<DataType> columnTypes;
final List<Function<Integer, Object>> valueFunction;
final String createTableSchema;
- final String expectedFailureMessage;
+ final String[] expectedFailureMessages;
final int numRows = 10_000;
Function<Object[], String> columnMapperValidation =
columns -> String.format(String.join(":", Collections.nCopies(columns.length, "%s")), columns);
@@ -492,7 +497,7 @@ class BulkWriteDataTypesTest extends SharedClusterSparkIntegrationTestBase
this.columnTypes = columnTypes;
this.valueFunction = valueFunction;
this.createTableSchema = createTableSchema;
- this.expectedFailureMessage = null;
+ this.expectedFailureMessages = null;
}
TypeTestSetup(String tableName,
@@ -500,14 +505,14 @@ class BulkWriteDataTypesTest extends SharedClusterSparkIntegrationTestBase
List<DataType> columnTypes,
List<Function<Integer, Object>> valueFunction,
String createTableSchema,
- String expectedFailureMessage)
+ String... expectedFailureMessages)
{
this.tableName = tableName;
this.columns = columns;
this.columnTypes = columnTypes;
this.valueFunction = valueFunction;
this.createTableSchema = createTableSchema;
- this.expectedFailureMessage = expectedFailureMessage;
+ this.expectedFailureMessages = expectedFailureMessages;
}
}
}
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkTestUtils.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkTestUtils.java
index bc9cdb74..0e3f8ef4 100644
--- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkTestUtils.java
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SparkTestUtils.java
@@ -52,6 +52,8 @@ import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SparkSession;
import org.jetbrains.annotations.NotNull;
+import static org.apache.cassandra.spark.bulkwriter.BulkSparkConf.JDK11_OPTIONS;
+import static org.apache.cassandra.spark.bulkwriter.BulkSparkConf.JDK17_OPTIONS;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
@@ -195,12 +197,16 @@ public class SparkTestUtils
// the quoted identifiers tests where we test mixed case
.set("spark.sql.caseSensitive", "True")
.set("spark.ui.enabled", "false")
+ .set("spark.driver.host", "127.0.0.1")
.set("spark.driver.bindAddress", "127.0.0.1")
.set("spark.master", "local[8,4]")
.set("spark.cassandra_analytics.cassandra.version", "5.0.0")
.set("spark.cassandra_analytics.sidecar.request.retries", "5")
.set("spark.cassandra_analytics.sidecar.request.retries.delay.milliseconds", "500")
- .set("spark.cassandra_analytics.sidecar.request.retries.max.delay.milliseconds", "500");
+ .set("spark.cassandra_analytics.sidecar.request.retries.max.delay.milliseconds", "500")
+ .set("spark.driver.extraJavaOptions", JDK11_OPTIONS + JDK17_OPTIONS)
+ .set("spark.executor.extraJavaOptions", JDK11_OPTIONS + JDK17_OPTIONS);
+
BulkSparkConf.setupSparkConf(sparkConf, true);
KryoRegister.setup(sparkConf);
return sparkConf;
diff --git a/cassandra-analytics-sidecar-client/build.gradle b/cassandra-analytics-sidecar-client/build.gradle
index bedf2c04..cad4d959 100644
--- a/cassandra-analytics-sidecar-client/build.gradle
+++ b/cassandra-analytics-sidecar-client/build.gradle
@@ -1,5 +1,3 @@
-import java.nio.file.Paths
-
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -29,8 +27,7 @@ configurations {
all*.exclude(group: 'log4j', module: 'log4j')
}
-if (propertyWithDefault("artifactType", null) == "common")
-{
+if (propertyWithDefault("artifactType", null) == "common") {
apply from: "$rootDir/gradle/common/publishing.gradle"
}
@@ -41,4 +38,4 @@ dependencies {
implementation "org.slf4j:slf4j-api:${slf4jApiVersion}"
api(project(path: ':analytics-sidecar-vertx-client-shaded', configuration: 'shadow'))
-}
\ No newline at end of file
+}
diff --git a/cassandra-analytics-spark-converter/src/main/java/org/apache/cassandra/spark/data/converter/types/StringFeatures.java b/cassandra-analytics-spark-converter/src/main/java/org/apache/cassandra/spark/data/converter/types/StringFeatures.java
index cc1e9a9d..a524cd19 100644
--- a/cassandra-analytics-spark-converter/src/main/java/org/apache/cassandra/spark/data/converter/types/StringFeatures.java
+++ b/cassandra-analytics-spark-converter/src/main/java/org/apache/cassandra/spark/data/converter/types/StringFeatures.java
@@ -79,6 +79,6 @@ interface StringFeatures extends SparkType
{
return STRING_COMPARATOR.compare(first.toString(), second.toString());
}
- return ((UTF8String) first).compare((UTF8String) second);
+ return ((UTF8String) first).compareTo((UTF8String) second);
}
}
diff --git a/cassandra-analytics-spark-converter/src/main/scala-2.11-spark-2/org/apache/cassandra/spark/utils/ScalaConversionUtils.java b/cassandra-analytics-spark-converter/src/main/scala-2.11-spark-2/org/apache/cassandra/spark/utils/ScalaConversionUtils.java
deleted file mode 100644
index 5cfb85bc..00000000
--- a/cassandra-analytics-spark-converter/src/main/scala-2.11-spark-2/org/apache/cassandra/spark/utils/ScalaConversionUtils.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.cassandra.spark.utils;
-
-import java.util.List;
-
-import scala.collection.JavaConversions;
-import scala.collection.mutable.Seq;
-
-/**
- * Compatibility layer for scala conversions
- */
-public final class ScalaConversionUtils
-{
- private ScalaConversionUtils()
- {
- throw new IllegalStateException(getClass() + " is static utility class and shall not be instantiated");
- }
-
- public static <A> java.lang.Iterable<A> asJavaIterable(scala.collection.Iterable<A> iterable)
- {
- return JavaConversions.asJavaIterable(iterable);
- }
-
- public static <A> scala.collection.Iterator<A> asScalaIterator(java.util.Iterator<A> iterator)
- {
- return JavaConversions.asScalaIterator(iterator);
- }
-
- public static <A, B> java.util.Map<A, B> mapAsJavaMap(scala.collection.Map<A, B> map)
- {
- return JavaConversions.mapAsJavaMap(map);
- }
-
- public static <A> List<A> mutableSeqAsJavaList(Seq<A> seq)
- {
- return JavaConversions.mutableSeqAsJavaList(seq);
- }
-}
diff --git a/cassandra-analytics-spark-converter/src/main/scala-2.11-spark-2/org/apache/cassandra/spark/utils/SparkTypeUtils.java b/cassandra-analytics-spark-converter/src/main/scala-2.11-spark-2/org/apache/cassandra/spark/utils/SparkTypeUtils.java
deleted file mode 100644
index 4df97e31..00000000
--- a/cassandra-analytics-spark-converter/src/main/scala-2.11-spark-2/org/apache/cassandra/spark/utils/SparkTypeUtils.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.cassandra.spark.utils;
-
-import java.util.Comparator;
-import java.util.concurrent.TimeUnit;
-
-import com.google.common.primitives.Ints;
-
-import org.apache.cassandra.bridge.type.InternalDuration;
-import org.apache.spark.unsafe.types.CalendarInterval;
-
-public final class SparkTypeUtils
-{
- private SparkTypeUtils()
- {
- throw new IllegalStateException(getClass() + " is static utility class and shall not be instantiated");
- }
-
- public static final Comparator<CalendarInterval> CALENDAR_INTERVAL_COMPARATOR =
- Comparator.<CalendarInterval>comparingInt(interval -> interval.months)
- .thenComparingLong(interval -> interval.microseconds);
-
- public static CalendarInterval convertDuration(InternalDuration duration)
- {
- // Unfortunately, it loses precision when converting to the spark data type.
- long micros = TimeUnit.NANOSECONDS.toMicros(duration.nanoseconds);
- micros += duration.days * CalendarInterval.MICROS_PER_DAY;
- return new CalendarInterval(duration.months, micros);
- }
-
- public static InternalDuration convertDuration(CalendarInterval interval)
- {
- int days = Ints.checkedCast(interval.microseconds / CalendarInterval.MICROS_PER_DAY);
- long microsRemain = interval.microseconds % CalendarInterval.MICROS_PER_DAY;
- return new InternalDuration(interval.months, days, TimeUnit.MICROSECONDS.toNanos(microsRemain));
- }
-}
diff --git a/cassandra-analytics-spark-converter/src/main/scala-2.12-spark-2/org/apache/cassandra/spark/utils/ScalaConversionUtils.java b/cassandra-analytics-spark-converter/src/main/scala-2.13-spark-4/org/apache/cassandra/spark/utils/ScalaConversionUtils.java
similarity index 75%
rename from cassandra-analytics-spark-converter/src/main/scala-2.12-spark-2/org/apache/cassandra/spark/utils/ScalaConversionUtils.java
rename to cassandra-analytics-spark-converter/src/main/scala-2.13-spark-4/org/apache/cassandra/spark/utils/ScalaConversionUtils.java
index 23eabb3f..e225e1ff 100644
--- a/cassandra-analytics-spark-converter/src/main/scala-2.12-spark-2/org/apache/cassandra/spark/utils/ScalaConversionUtils.java
+++ b/cassandra-analytics-spark-converter/src/main/scala-2.13-spark-4/org/apache/cassandra/spark/utils/ScalaConversionUtils.java
@@ -21,8 +21,8 @@ package org.apache.cassandra.spark.utils;
import java.util.List;
-import scala.collection.JavaConverters;
import scala.collection.mutable.Seq;
+import scala.jdk.javaapi.CollectionConverters;
/**
* Compatibility layer for scala conversions
@@ -34,23 +34,23 @@ public final class ScalaConversionUtils
throw new IllegalStateException(getClass() + " is static utility class and shall not be instantiated");
}
- public static <A> java.lang.Iterable<A> asJavaIterable(scala.collection.Iterable<A> iterable)
+ public static <A> Iterable<A> asJavaIterable(scala.collection.Iterable<A> iterable)
{
- return JavaConverters.asJavaIterable(iterable);
+ return CollectionConverters.asJava(iterable);
}
public static <A> scala.collection.Iterator<A> asScalaIterator(java.util.Iterator<A> iterator)
{
- return JavaConverters.asScalaIterator(iterator);
+ return CollectionConverters.asScala(iterator);
}
- public static <A, B> java.util.Map<A, B> mapAsJavaMap(scala.collection.Map<A, B> map)
+ public static <K, V> java.util.Map<K, V> mapAsJavaMap(scala.collection.Map<K, V> map)
{
- return JavaConverters.mapAsJavaMap(map);
+ return CollectionConverters.asJava(map);
}
public static <A> List<A> mutableSeqAsJavaList(Seq<A> seq)
{
- return JavaConverters.mutableSeqAsJavaList(seq);
+ return CollectionConverters.asJava(seq);
}
}
diff --git a/cassandra-analytics-spark-converter/src/main/scala-2.12-spark-2/org/apache/cassandra/spark/utils/SparkTypeUtils.java b/cassandra-analytics-spark-converter/src/main/scala-2.13-spark-4/org/apache/cassandra/spark/utils/SparkTypeUtils.java
similarity index 76%
rename from cassandra-analytics-spark-converter/src/main/scala-2.12-spark-2/org/apache/cassandra/spark/utils/SparkTypeUtils.java
rename to cassandra-analytics-spark-converter/src/main/scala-2.13-spark-4/org/apache/cassandra/spark/utils/SparkTypeUtils.java
index 4df97e31..4d200b35 100644
--- a/cassandra-analytics-spark-converter/src/main/scala-2.12-spark-2/org/apache/cassandra/spark/utils/SparkTypeUtils.java
+++ b/cassandra-analytics-spark-converter/src/main/scala-2.13-spark-4/org/apache/cassandra/spark/utils/SparkTypeUtils.java
@@ -22,8 +22,6 @@ package org.apache.cassandra.spark.utils;
import java.util.Comparator;
import java.util.concurrent.TimeUnit;
-import com.google.common.primitives.Ints;
-
import org.apache.cassandra.bridge.type.InternalDuration;
import org.apache.spark.unsafe.types.CalendarInterval;
@@ -36,20 +34,17 @@ public final class SparkTypeUtils
public static final Comparator<CalendarInterval> CALENDAR_INTERVAL_COMPARATOR =
Comparator.<CalendarInterval>comparingInt(interval -> interval.months)
+ .thenComparingInt(interval -> interval.days)
.thenComparingLong(interval -> interval.microseconds);
public static CalendarInterval convertDuration(InternalDuration duration)
{
// Unfortunately, it loses precision when converting to the spark data type.
- long micros = TimeUnit.NANOSECONDS.toMicros(duration.nanoseconds);
- micros += duration.days * CalendarInterval.MICROS_PER_DAY;
- return new CalendarInterval(duration.months, micros);
+ return new CalendarInterval(duration.months, duration.days, TimeUnit.NANOSECONDS.toMicros(duration.nanoseconds));
}
public static InternalDuration convertDuration(CalendarInterval interval)
{
- int days = Ints.checkedCast(interval.microseconds / CalendarInterval.MICROS_PER_DAY);
- long microsRemain = interval.microseconds % CalendarInterval.MICROS_PER_DAY;
- return new InternalDuration(interval.months, days, TimeUnit.MICROSECONDS.toNanos(microsRemain));
+ return new InternalDuration(interval.months, interval.days, TimeUnit.MICROSECONDS.toNanos(interval.microseconds));
}
}
diff --git a/cassandra-five-zero-bridge/build.gradle b/cassandra-five-zero-bridge/build.gradle
index 31882e13..9a785d3f 100644
--- a/cassandra-five-zero-bridge/build.gradle
+++ b/cassandra-five-zero-bridge/build.gradle
@@ -23,6 +23,8 @@ plugins {
id('java-library')
}
+apply from: "${project.rootDir}/gradle/common/javaOptions.gradle"
+
configurations {
all*.exclude(group: 'org.slf4j', module: 'slf4j-log4j12')
all*.exclude(group: 'log4j', module: 'log4j')
@@ -107,26 +109,8 @@ test {
// automatically run BIG and BTI tests for Cassandra 5.x bridge
tasks.register('testBti', Test) {
doFirst {
- if (JavaVersion.current().isJava11Compatible()) {
- def JDK11_OPTIONS = ['-Djdk.attach.allowAttachSelf=true',
- '--add-exports', 'java.base/jdk.internal.misc=ALL-UNNAMED',
- '--add-exports', 'java.base/jdk.internal.ref=ALL-UNNAMED',
- '--add-exports', 'java.base/sun.nio.ch=ALL-UNNAMED',
- '--add-exports', 'java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED',
- '--add-exports', 'java.rmi/sun.rmi.registry=ALL-UNNAMED',
- '--add-exports', 'java.rmi/sun.rmi.server=ALL-UNNAMED',
- '--add-exports', 'java.sql/java.sql=ALL-UNNAMED',
- '--add-opens', 'java.base/java.lang.module=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.loader=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.ref=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.reflect=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.math=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.module=ALL-UNNAMED',
- '--add-opens', 'java.base/jdk.internal.util.jar=ALL-UNNAMED',
- '--add-opens', 'jdk.management/com.sun.management.internal=ALL-UNNAMED']
- jvmArgs(JDK11_OPTIONS)
- println("JVM arguments for $project.name are $allJvmArgs")
- }
+ jvmArgs(project.ext.JDK_OPTIONS)
+ println("JVM arguments for $project.name are $allJvmArgs")
}
systemProperty "cassandra.analytics.bridges.sstable_format", "bti"
@@ -146,4 +130,4 @@ tasks.register('testBti', Test) {
check {
dependsOn(tasks.testBti)
-}
\ No newline at end of file
+}
diff --git a/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java
index 4d55d81d..b6c9853a 100644
--- a/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java
+++ b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java
@@ -23,12 +23,12 @@ import java.io.IOException;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.Collection;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Stream;
import com.google.common.collect.ImmutableMultimap;
import com.google.common.collect.Multimap;
-import org.apache.commons.lang.mutable.MutableInt;
-import org.apache.commons.lang.mutable.MutableLong;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -119,8 +119,8 @@ public class IndexOffsetTests
LOGGER.info("Testing index offsets numKeys={} sparkPartitions={} partitioner={} enableCompression={}",
numKeys, ranges.size(), partitioner.name(), enableCompression);
- MutableInt skippedPartitions = new MutableInt(0);
- MutableLong skippedDataOffsets = new MutableLong(0);
+ AtomicInteger skippedPartitions = new AtomicInteger(0);
+ AtomicLong skippedDataOffsets = new AtomicLong(0);
int[][] counts = new int[numPartitions][numRowsPerPartition];
for (TokenRange range : ranges)
{
@@ -130,12 +130,12 @@ public class IndexOffsetTests
{
public void skippedPartition(ByteBuffer key, BigInteger token)
{
- skippedPartitions.add(1);
+ skippedPartitions.addAndGet(1);
}
public void skippedDataDbStartOffset(long length)
{
- skippedDataOffsets.add(length);
+ skippedDataOffsets.addAndGet(length);
}
})
.build();
diff --git a/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java
index fd35c2ca..18821a2b 100644
--- a/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java
+++ b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java
@@ -45,7 +45,6 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.google.common.collect.ImmutableMap;
-import org.apache.commons.lang.StringUtils;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -562,7 +561,7 @@ public class SSTableReaderTests
colBuf.get();
String colName = ByteBufferUtils.string(ByteBufferUtils.readBytesWithShortLength(colBuf));
colBuf.get();
- if (StringUtils.isEmpty(colName))
+ if (colName == null || colName.isEmpty())
{
continue;
}
@@ -942,7 +941,7 @@ public class SSTableReaderTests
colBuf.get();
String colName = ByteBufferUtils.string(ByteBufferUtils.readBytesWithShortLength(colBuf));
colBuf.get();
- if (StringUtils.isEmpty(colName))
+ if (colName == null || colName.isEmpty())
{
continue;
}
diff --git a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java
index b6d7db8c..84fe9193 100644
--- a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java
+++ b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java
@@ -23,11 +23,11 @@ import java.io.IOException;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.Collection;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import com.google.common.collect.ImmutableMultimap;
import com.google.common.collect.Multimap;
-import org.apache.commons.lang.mutable.MutableInt;
-import org.apache.commons.lang.mutable.MutableLong;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -111,8 +111,8 @@ public class IndexOffsetTests
LOGGER.info("Testing index offsets numKeys={} sparkPartitions={} partitioner={} enableCompression={}",
numKeys, ranges.size(), partitioner.name(), enableCompression);
- MutableInt skippedPartitions = new MutableInt(0);
- MutableLong skippedDataOffsets = new MutableLong(0);
+ AtomicInteger skippedPartitions = new AtomicInteger(0);
+ AtomicLong skippedDataOffsets = new AtomicLong(0);
int[] counts = new int[numKeys];
for (TokenRange range : ranges)
{
@@ -122,12 +122,12 @@ public class IndexOffsetTests
{
public void skippedPartition(ByteBuffer key, BigInteger token)
{
- skippedPartitions.add(1);
+ skippedPartitions.addAndGet(1);
}
public void skippedDataDbStartOffset(long length)
{
- skippedDataOffsets.add(length);
+ skippedDataOffsets.addAndGet(length);
}
})
.build();
diff --git a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java
index d3b4ac38..1f8c7ef6 100644
--- a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java
+++ b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java
@@ -46,7 +46,6 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.google.common.collect.ImmutableMap;
-import org.apache.commons.lang.StringUtils;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -536,7 +535,7 @@ public class SSTableReaderTests
colBuf.get();
String colName = ByteBufferUtils.string(ByteBufferUtils.readBytesWithShortLength(colBuf));
colBuf.get();
- if (StringUtils.isEmpty(colName))
+ if (colName == null || colName.isEmpty())
{
continue;
}
@@ -916,7 +915,7 @@ public class SSTableReaderTests
colBuf.get();
String colName = ByteBufferUtils.string(ByteBufferUtils.readBytesWithShortLength(colBuf));
colBuf.get();
- if (StringUtils.isEmpty(colName))
+ if (colName == null || colName.isEmpty())
{
continue;
}
diff --git a/gradle.properties b/gradle.properties
index d19e9a51..30e1aa3b 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -20,7 +20,7 @@ version=0.5-SNAPSHOT
snapshot=true
description=Apache Cassandra Analytics
-analyticsJDKLevel=11
+analyticsJDKLevel=17
cassandra40Version=4.0.17
cassandra50Version=5.0.5
sidecarVersion=0.2.0
@@ -30,8 +30,8 @@ assertjCoreVersion=3.24.2
quickTheoriesVersion=0.26
mockitoVersion=3.12.4
jnaVersion=5.9.0
-scala=2.12
-spark=3
+scala=2.13
+spark=4
kryoVersion=4.0.2
slf4jApiVersion=1.7.30
guavaVersion=16.0.1
diff --git a/gradle/common/java11Options.gradle b/gradle/common/java11Options.gradle
new file mode 100644
index 00000000..ed427c61
--- /dev/null
+++ b/gradle/common/java11Options.gradle
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+project.ext.JDK11_OPTIONS = [
+ '-Djdk.attach.allowAttachSelf=true',
+ '--add-exports', 'java.base/jdk.internal.misc=ALL-UNNAMED',
+ '--add-exports', 'java.base/jdk.internal.ref=ALL-UNNAMED',
+ '--add-exports', 'java.base/sun.nio.ch=ALL-UNNAMED',
+ '--add-exports', 'java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED',
+ '--add-exports', 'java.rmi/sun.rmi.registry=ALL-UNNAMED',
+ '--add-exports', 'java.rmi/sun.rmi.server=ALL-UNNAMED',
+ '--add-exports', 'java.sql/java.sql=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.lang.module=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.loader=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.ref=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.reflect=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.math=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.module=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.util.jar=ALL-UNNAMED',
+ '--add-opens', 'jdk.management/com.sun.management.internal=ALL-UNNAMED',
+]
diff --git a/gradle/common/java17Options.gradle b/gradle/common/java17Options.gradle
new file mode 100644
index 00000000..d0c7bec1
--- /dev/null
+++ b/gradle/common/java17Options.gradle
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+
+
+project.ext.JDK17_OPTIONS = [
+ '--add-exports', 'java.rmi/sun.rmi.transport=ALL-UNNAMED',
+ '--add-exports', 'java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.io=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.lang=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.lang.invoke=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.lang.reflect=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.math=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.net=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.nio=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.util=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.util.concurrent=ALL-UNNAMED',
+ '--add-opens', 'java.base/java.util.concurrent.atomic=ALL-UNNAMED',
+ '--add-opens', 'java.base/jdk.internal.misc=ALL-UNNAMED',
+ '--add-opens', 'java.base/sun.nio.ch=ALL-UNNAMED',
+ '--add-opens', 'java.base/sun.util.calendar=ALL-UNNAMED',
+ '--add-opens', 'java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED',
+]
diff --git a/gradle/common/javaOptions.gradle b/gradle/common/javaOptions.gradle
new file mode 100644
index 00000000..434868bc
--- /dev/null
+++ b/gradle/common/javaOptions.gradle
@@ -0,0 +1,26 @@
+/*
+ * 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.
+ */
+
+apply from: "${project.rootDir}/gradle/common/java11Options.gradle"
+apply from: "${project.rootDir}/gradle/common/java17Options.gradle"
+
+project.ext.JDK_OPTIONS = project.ext.JDK11_OPTIONS
+
+if (JavaVersion.current().isCompatibleWith(JavaVersion.VERSION_17)) {
+ project.ext.JDK_OPTIONS += project.ext.JDK17_OPTIONS
+}
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
index ca025c83..d4081da4 100644
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -1,6 +1,6 @@
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.14-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.14.3-bin.zip
networkTimeout=10000
validateDistributionUrl=true
zipStoreBase=GRADLE_USER_HOME
diff --git a/cassandra-analytics-sidecar-client/build.gradle b/profiles/scala-2.13-spark-4-jdk-17.gradle
similarity index 56%
copy from cassandra-analytics-sidecar-client/build.gradle
copy to profiles/scala-2.13-spark-4-jdk-17.gradle
index bedf2c04..859cb6b2 100644
--- a/cassandra-analytics-sidecar-client/build.gradle
+++ b/profiles/scala-2.13-spark-4-jdk-17.gradle
@@ -1,5 +1,3 @@
-import java.nio.file.Paths
-
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -19,26 +17,12 @@ import java.nio.file.Paths
* under the License.
*/
-plugins {
- id('java-library')
- id('maven-publish')
-}
-
-configurations {
- all*.exclude(group: 'org.slf4j', module: 'slf4j-log4j12')
- all*.exclude(group: 'log4j', module: 'log4j')
-}
-
-if (propertyWithDefault("artifactType", null) == "common")
-{
- apply from: "$rootDir/gradle/common/publishing.gradle"
+ext {
+ jacksonVersion="2.18.2"
+ jacksonScalaModuleVersion="2.18.2"
+ scalaMajorVersion="2.13"
+ scalaVersion="2.13.16"
+ sparkGroupId="org.apache.spark"
+ sparkMajorVersion="4"
+ sparkVersion="4.0.1"
}
-
-dependencies {
- compileOnly("com.google.guava:guava:${guavaVersion}")
- implementation project(":cassandra-analytics-common")
-
- implementation "org.slf4j:slf4j-api:${slf4jApiVersion}"
-
- api(project(path: ':analytics-sidecar-vertx-client-shaded', configuration: 'shadow'))
-}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org