This is an automated email from the ASF dual-hosted git repository.
dcapwell pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cassandra-easy-stress.git
The following commit(s) were added to refs/heads/main by this push:
new e4ca160 Make sure the parquet file is compressed and uses the v2
format (#109)
e4ca160 is described below
commit e4ca160d6183a08157ad9f118a9d2136b211836d
Author: dcapwell <[email protected]>
AuthorDate: Tue Oct 20 23:18:57 2026 -0700
Make sure the parquet file is compressed and uses the v2 format (#109)
---
CLAUDE.md | 10 ++++
build.gradle.kts | 8 +++
.../easystress/collector/ParquetCollector.kt | 13 ++++-
.../easystress/collector/ParquetCollectorTest.kt | 63 ++++++++++++++++++++++
4 files changed, 93 insertions(+), 1 deletion(-)
diff --git a/CLAUDE.md b/CLAUDE.md
index c7af45e..e63ce2c 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -13,6 +13,16 @@ For comprehensive testing documentation, see the [Testing
section in README.md](
- Format code: `./gradlew ktlintFormat`
- Check formatting: `./gradlew ktlintCheck`
- Generate docs: `./gradlew docs`
+- Fast pre-PR checks: `./gradlew preflight` (ktlintCheck, detekt, test,
distTar)
+
+## Before declaring work done (required)
+
+Run `./gradlew ktlintFormat` then `./gradlew preflight` and fix any failures.
This covers
+the cheap CI jobs: ktlint, detekt, unit tests, and `distTar`. The slower
per-version matrix
+(test40/test41/test50 on JDK 21 and 25) is left to CI; no need to run it
locally.
+
+When adding a dependency, confirm it is the maintained artifact and that it
lands in the
+distribution (`distTar`).
## Code Style Guidelines
- Kotlin version: 1.9.0
diff --git a/build.gradle.kts b/build.gradle.kts
index 2d38a58..5ae2508 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -351,3 +351,11 @@ tasks.withType<DetektCreateBaselineTask>().configureEach {
tasks.withType<ShadowJar>().configureEach {
isZip64 = true
}
+
+// Fast local checks mirroring the cheap CI jobs (lint, detekt, unit tests,
distTar).
+// The slower per-version matrix (test40/test41/test50) is left to CI.
+tasks.register("preflight") {
+ group = "verification"
+ description = "Run fast CI-equivalent checks before sending a PR"
+ dependsOn("ktlintCheck", "detekt", "test", "distTar")
+}
diff --git
a/src/main/kotlin/org/apache/cassandra/easystress/collector/ParquetCollector.kt
b/src/main/kotlin/org/apache/cassandra/easystress/collector/ParquetCollector.kt
index 1b54476..959681b 100644
---
a/src/main/kotlin/org/apache/cassandra/easystress/collector/ParquetCollector.kt
+++
b/src/main/kotlin/org/apache/cassandra/easystress/collector/ParquetCollector.kt
@@ -4,10 +4,12 @@ import com.datastax.oss.driver.api.core.cql.AsyncResultSet
import com.google.common.base.Throwables
import org.apache.cassandra.easystress.Either
import org.apache.hadoop.fs.Path
+import org.apache.parquet.column.ParquetProperties
import org.apache.parquet.example.data.Group
import org.apache.parquet.example.data.simple.SimpleGroupFactory
import org.apache.parquet.hadoop.ParquetWriter
import org.apache.parquet.hadoop.example.ExampleParquetWriter
+import org.apache.parquet.hadoop.metadata.CompressionCodecName
import org.apache.parquet.schema.MessageTypeParser
import java.io.File
import java.util.concurrent.TimeUnit
@@ -46,7 +48,16 @@ class ParquetCollector(
"}",
)
groupFactory = SimpleGroupFactory(schema)
- writer =
ExampleParquetWriter.builder(path).withType(schema).build()
+ // Defaults are UNCOMPRESSED + PARQUET_1_0 (PLAIN int64), which
for long runs produces multi-GB files.
+ // ZSTD + PARQUET_2_0 (DELTA_BINARY_PACKED for the time/duration
columns) is ~3x smaller with the same
+ // schema and values.
+ writer =
+ ExampleParquetWriter
+ .builder(path)
+ .withType(schema)
+ .withCompressionCodec(CompressionCodecName.ZSTD)
+
.withWriterVersion(ParquetProperties.WriterVersion.PARQUET_2_0)
+ .build()
}
override fun write(event: Event) {
diff --git
a/src/test/kotlin/org/apache/cassandra/easystress/collector/ParquetCollectorTest.kt
b/src/test/kotlin/org/apache/cassandra/easystress/collector/ParquetCollectorTest.kt
index 6b7824b..2b97ac7 100644
---
a/src/test/kotlin/org/apache/cassandra/easystress/collector/ParquetCollectorTest.kt
+++
b/src/test/kotlin/org/apache/cassandra/easystress/collector/ParquetCollectorTest.kt
@@ -1,12 +1,24 @@
package org.apache.cassandra.easystress.collector
import com.datastax.oss.driver.api.core.CqlSession
+import com.datastax.oss.driver.api.core.cql.AsyncResultSet
+import com.datastax.oss.driver.api.core.cql.BoundStatement
import com.google.common.io.Files
import io.mockk.mockk
import org.apache.cassandra.easystress.Context
+import org.apache.cassandra.easystress.Either
import org.apache.cassandra.easystress.Metrics
import org.apache.cassandra.easystress.commands.Run
import org.apache.cassandra.easystress.generators.Registry
+import org.apache.cassandra.easystress.workloads.Operation
+import org.apache.hadoop.conf.Configuration
+import org.apache.parquet.column.Encoding
+import org.apache.parquet.example.data.Group
+import org.apache.parquet.hadoop.ParquetFileReader
+import org.apache.parquet.hadoop.ParquetReader
+import org.apache.parquet.hadoop.example.GroupReadSupport
+import org.apache.parquet.hadoop.metadata.CompressionCodecName
+import org.apache.parquet.hadoop.util.HadoopInputFile
import org.assertj.core.api.Assertions
import org.assertj.core.api.Condition
import org.junit.jupiter.api.Test
@@ -51,6 +63,57 @@ internal class ParquetCollectorTest {
)
}
+ @Test
+ fun usesZstdAndV2Encodings() {
+ val file = tempDir!!.resolve("rawlog.parquet").toFile()
+ val op = Operation.SelectStatement(mockk<BoundStatement>())
+ val writer = ParquetCollector.ParquetTableWriter(file)
+ val rows = 100
+ for (i in 0 until rows) {
+ writer.write(
+ AsyncCollector.Event(
+ op,
+ Either.Left(mockk<AsyncResultSet>()),
+ op.createdAtNanos + i,
+ op.createdAtNanos + 1_000 + i * 2L,
+ ),
+ )
+ }
+ writer.close()
+
+ val path =
+ org.apache.hadoop.fs
+ .Path(file.toURI())
+ val conf = Configuration()
+ ParquetFileReader.open(HadoopInputFile.fromPath(path, conf)).use {
reader ->
+ val columns = reader.footer.blocks.flatMap { it.columns }
+ Assertions.assertThat(columns).isNotEmpty
+ Assertions.assertThat(columns.map { it.codec
}).containsOnly(CompressionCodecName.ZSTD)
+ val int64Encodings =
+ columns
+ .filter { it.path.toDotString().endsWith("_ms") ||
it.path.toDotString().endsWith("_ns") }
+ .flatMap { it.encodings }
+
Assertions.assertThat(int64Encodings).contains(Encoding.DELTA_BINARY_PACKED)
+ }
+
+ // semantics must not change: same columns, same values
+ val read = mutableListOf<Group>()
+ ParquetReader.builder(GroupReadSupport(),
path).withConf(conf).build().use { reader ->
+ while (true) {
+ read.add(reader.read() ?: break)
+ }
+ }
+ Assertions.assertThat(read).hasSize(rows)
+ read.forEachIndexed { i, g ->
+ Assertions.assertThat(g.getString("operation",
0)).isEqualTo("Select")
+ Assertions.assertThat(g.getBoolean("success", 0)).isTrue
+ Assertions.assertThat(g.getString("failure_reason", 0)).isEmpty()
+ Assertions.assertThat(g.getLong("request_start_time_ms",
0)).isEqualTo(op.createdAtMillis)
+ Assertions.assertThat(g.getLong("request_duration_ns",
0)).isEqualTo(1_000 + i * 2L)
+ Assertions.assertThat(g.getLong("service_duration_ns",
0)).isEqualTo(1_000 + i * 2L - i)
+ }
+ }
+
companion object {
val ctx =
Context(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
