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 46c35d0 CASSANDRA-19411: Bulk reader fails to produce a row when
regular column values are null
46c35d0 is described below
commit 46c35d0ef2efb66512133a7913df9936b0a80dc8
Author: Francisco Guerrero <fran...@apache.org>
AuthorDate: Mon Feb 19 20:50:16 2024 -0800
CASSANDRA-19411: Bulk reader fails to produce a row when regular column
values are null
Bulk Reader won't emit a row when the regular column values are all `null`.
For example,
a schema `PK` = `a`, `b` ; `CK` = `c`, `d` ; and columns = `e`, `f`.
| a | b | c | d | e | f |
| --- | --- | --- | --- | ---- | ---- |
| pk1 | pk2 | ck1 | ck2 | null | null |
When queried from Analytics bulk reader, it won't produce a row.
This issue also occurs when the projected regular column values are all
`null`, where
other non-projected columns might have some values.
Patch by Francisco Guerrero; Reviewed by Yifan Cai for CASSANDRA-19411
---
CHANGES.txt | 1 +
.../spark/sparksql/AbstractSparkRowIterator.java | 6 +-
.../cassandra/spark/sparksql/FullRowBuilder.java | 12 +-
.../spark/sparksql/SparkCellIterator.java | 51 +++++---
.../cassandra/spark/sparksql/SparkRowIterator.java | 2 +-
.../cassandra/spark/sparksql/SparkRowIterator.java | 16 +--
.../org/apache/cassandra/spark/EndToEndTests.java | 113 ++++++++++++++++
.../java/org/apache/cassandra/spark/Tester.java | 143 +++++++++------------
.../cassandra/spark/reader/EmptyStreamScanner.java | 6 +
.../cassandra/spark/reader/StreamScanner.java | 5 +
.../spark/reader/common/IndexIterator.java | 6 +
.../cassandra/spark/utils/test/TestSchema.java | 28 +++-
.../spark/reader/AbstractStreamScanner.java | 6 +
13 files changed, 271 insertions(+), 124 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 9daa5f0..1472baf 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
1.0.0
+ * Bulk reader fails to produce a row when regular column values are null
(CASSANDRA-19411)
* Use XXHash32 for digest calculation of SSTables (CASSANDRA-19369)
* Startup Validation Failures when Checking Sidecar Connectivity
(CASSANDRA-19377)
* No longer need to synchronize on Schema.instance after Cassandra 4.0.12
(CASSANDRA-19351)
diff --git
a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/AbstractSparkRowIterator.java
b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/AbstractSparkRowIterator.java
index 7882fe9..cc2f810 100644
---
a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/AbstractSparkRowIterator.java
+++
b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/AbstractSparkRowIterator.java
@@ -49,8 +49,8 @@ abstract class AbstractSparkRowIterator
protected final List<SchemaFeature> requestedFeatures;
protected final CqlTable cqlTable;
- protected final boolean noValueColumns;
protected final StructType columnFilter;
+ protected final boolean hasProjectedValueColumns;
private Cell cell = null;
private InternalRow row = null;
@@ -67,7 +67,7 @@ abstract class AbstractSparkRowIterator
this.stats.openedSparkRowIterator();
this.openTimeNanos = System.nanoTime();
this.requestedFeatures = dataLayer.requestedFeatures();
- this.noValueColumns = it.noValueColumns();
+ this.hasProjectedValueColumns = it.hasProjectedValueColumns();
this.builder = newBuilder();
}
@@ -132,7 +132,7 @@ abstract class AbstractSparkRowIterator
builder.onCell(cell);
- if (!noValueColumns)
+ if (hasProjectedValueColumns)
{
// If schema has value column
builder.copyValue(cell);
diff --git
a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/FullRowBuilder.java
b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/FullRowBuilder.java
index 823fdbe..5a06343 100644
---
a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/FullRowBuilder.java
+++
b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/FullRowBuilder.java
@@ -35,18 +35,18 @@ class FullRowBuilder implements RowBuilder
static final Object[] EMPTY_RESULT = new Object[0];
final int numColumns;
final int numCells;
- final boolean noValueColumns;
+ final boolean hasProjectedValueColumns;
int extraColumns;
Object[] result;
int count;
private final CqlTable cqlTable;
- FullRowBuilder(CqlTable cqlTable, boolean noValueColumns)
+ FullRowBuilder(CqlTable cqlTable, boolean hasProjectedValueColumns)
{
this.cqlTable = cqlTable;
this.numColumns = cqlTable.numFields();
- this.numCells = cqlTable.numNonValueColumns() + (noValueColumns ? 0 :
1);
- this.noValueColumns = noValueColumns;
+ this.hasProjectedValueColumns = hasProjectedValueColumns;
+ this.numCells = cqlTable.numNonValueColumns() +
(hasProjectedValueColumns ? 1 : 0);
}
@Override
@@ -80,7 +80,7 @@ class FullRowBuilder implements RowBuilder
public void copyKeys(Cell cell)
{
// Need to handle special case where schema is only partition or
clustering keys - i.e. no value columns
- int length = noValueColumns ? cell.values.length : cell.values.length
- 1;
+ int length = !hasProjectedValueColumns ? cell.values.length :
cell.values.length - 1;
System.arraycopy(cell.values, 0, result, 0, length);
count += length;
}
@@ -108,7 +108,7 @@ class FullRowBuilder implements RowBuilder
@Override
public boolean hasRegularValueColumn()
{
- return !noValueColumns;
+ return hasProjectedValueColumns;
}
@Override
diff --git
a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/SparkCellIterator.java
b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/SparkCellIterator.java
index c054ebc..58cb424 100644
---
a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/SparkCellIterator.java
+++
b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/sparksql/SparkCellIterator.java
@@ -25,7 +25,6 @@ import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
-import java.util.Set;
import java.util.stream.Collectors;
import org.slf4j.Logger;
@@ -57,7 +56,6 @@ public class SparkCellIterator implements Iterator<Cell>,
AutoCloseable
private final Stats stats;
private final CqlTable cqlTable;
private final Object[] values;
- private final boolean noValueColumns;
@Nullable
protected final PruneColumnFilter columnFilter;
private final long startTimeNanos;
@@ -74,6 +72,8 @@ public class SparkCellIterator implements Iterator<Cell>,
AutoCloseable
private long previousTimeNanos;
protected final int partitionId;
+ protected final int firstProjectedValueColumnPositionOrZero;
+ protected final boolean hasProjectedValueColumns;
public SparkCellIterator(int partitionId,
@NotNull DataLayer dataLayer,
@@ -88,19 +88,16 @@ public class SparkCellIterator implements Iterator<Cell>,
AutoCloseable
if (columnFilter != null)
{
LOGGER.info("Adding prune column filter columns='{}'",
String.join(",", columnFilter.requiredColumns()));
-
- // If we are reading only partition/clustering keys or static
columns, no value columns
- Set<String> valueColumns =
cqlTable.valueColumns().stream().map(CqlField::name).collect(Collectors.toSet());
- noValueColumns =
columnFilter.requiredColumns().stream().noneMatch(valueColumns::contains);
- }
- else
- {
- noValueColumns = cqlTable.numValueColumns() == 0;
}
+ hasProjectedValueColumns = cqlTable.numValueColumns() > 0 &&
+ cqlTable.valueColumns()
+ .stream()
+ .anyMatch(field -> columnFilter ==
null || columnFilter.requiredColumns().contains(field.name()));
+
// The value array copies across all the partition/clustering/static
columns
// and the single column value for this cell to the SparkRowIterator
- values = new Object[cqlTable.numNonValueColumns() + (noValueColumns ?
0 : 1)];
+ values = new Object[cqlTable.numNonValueColumns() +
(hasProjectedValueColumns ? 1 : 0)];
// Open compaction scanner
startTimeNanos = System.nanoTime();
@@ -111,6 +108,7 @@ public class SparkCellIterator implements Iterator<Cell>,
AutoCloseable
stats.openedCompactionScanner(openTimeNanos);
rid = scanner.rid();
stats.openedSparkCellIterator();
+ firstProjectedValueColumnPositionOrZero =
maybeGetPositionOfFirstProjectedValueColumnOrZero();
}
protected StreamScanner<Rid> openScanner(int partitionId,
@@ -130,9 +128,9 @@ public class SparkCellIterator implements Iterator<Cell>,
AutoCloseable
: null;
}
- public boolean noValueColumns()
+ public boolean hasProjectedValueColumns()
{
- return noValueColumns;
+ return hasProjectedValueColumns;
}
@Override
@@ -196,10 +194,18 @@ public class SparkCellIterator implements Iterator<Cell>,
AutoCloseable
String columnName = component != null ?
ByteBufferUtils.stringThrowRuntime(component) : null;
if (columnName == null || columnName.isEmpty())
{
- if (noValueColumns)
+ if (!hasProjectedValueColumns || !scanner.hasMoreColumns())
{
- // Special case where schema consists only of partition
keys, clustering keys or static columns, no value columns
- next = new Cell(values, 0, newRow, rid.getTimestamp());
+ if (hasProjectedValueColumns)
+ {
+ // null out the value of the cell for the case where
we have projected value columns
+ values[values.length - 1] = null;
+ }
+ // We use the position of a cell for a value column that
is projected, or zero if no value
+ // columns are projected. The column we find is irrelevant
because if we fall under this
+ // condition it means that we are in a situation where the
row has only PK + CK, but no
+ // regular columns.
+ next = new Cell(values,
firstProjectedValueColumnPositionOrZero, newRow, rid.getTimestamp());
return true;
}
@@ -359,4 +365,17 @@ public class SparkCellIterator implements Iterator<Cell>,
AutoCloseable
stats.fieldDeserialization(field, System.nanoTime() - now);
return value;
}
+
+ private int maybeGetPositionOfFirstProjectedValueColumnOrZero()
+ {
+ // find the position of the first value column that is projected
+ for (CqlField valueColumn : cqlTable.valueColumns())
+ {
+ if (columnFilter == null ||
columnFilter.includeColumn(valueColumn.name()))
+ {
+ return valueColumn.position();
+ }
+ }
+ return 0;
+ }
}
diff --git
a/cassandra-analytics-core/src/main/spark2/org/apache/cassandra/spark/sparksql/SparkRowIterator.java
b/cassandra-analytics-core/src/main/spark2/org/apache/cassandra/spark/sparksql/SparkRowIterator.java
index 26e560e..d77d980 100644
---
a/cassandra-analytics-core/src/main/spark2/org/apache/cassandra/spark/sparksql/SparkRowIterator.java
+++
b/cassandra-analytics-core/src/main/spark2/org/apache/cassandra/spark/sparksql/SparkRowIterator.java
@@ -56,7 +56,7 @@ public class SparkRowIterator extends
AbstractSparkRowIterator implements InputP
@NotNull
RowBuilder newBuilder()
{
- RowBuilder builder = new FullRowBuilder(cqlTable, noValueColumns);
+ RowBuilder builder = new FullRowBuilder(cqlTable,
hasProjectedValueColumns);
for (SchemaFeature feature : requestedFeatures)
{
builder = feature.decorate(builder);
diff --git
a/cassandra-analytics-core/src/main/spark3/org/apache/cassandra/spark/sparksql/SparkRowIterator.java
b/cassandra-analytics-core/src/main/spark3/org/apache/cassandra/spark/sparksql/SparkRowIterator.java
index 118574d..c4c786b 100644
---
a/cassandra-analytics-core/src/main/spark3/org/apache/cassandra/spark/sparksql/SparkRowIterator.java
+++
b/cassandra-analytics-core/src/main/spark3/org/apache/cassandra/spark/sparksql/SparkRowIterator.java
@@ -67,12 +67,12 @@ public class SparkRowIterator extends
AbstractSparkRowIterator implements Partit
String[] fieldNames = null;
if (columnFilter != null)
{
- builder = new PartialRowBuilder(columnFilter, cqlTable,
noValueColumns);
+ builder = new PartialRowBuilder(columnFilter, cqlTable,
hasProjectedValueColumns);
fieldNames = columnFilter.fieldNames();
}
else
{
- builder = new FullRowBuilder(cqlTable, noValueColumns);
+ builder = new FullRowBuilder(cqlTable, hasProjectedValueColumns);
}
for (SchemaFeature feature : requestedFeatures)
@@ -102,17 +102,17 @@ public class SparkRowIterator extends
AbstractSparkRowIterator implements Partit
PartialRowBuilder(@NotNull StructType requiredSchema,
CqlTable table,
- boolean noValueColumns)
+ boolean hasProjectedValueColumns)
{
- super(table, noValueColumns);
+ super(table, hasProjectedValueColumns);
this.requiredSchema = requiredSchema;
Set<String> requiredColumns =
Arrays.stream(requiredSchema.fields())
.map(StructField::name)
.collect(Collectors.toSet());
hasAllNonValueColumns = table.fields().stream()
-
.filter(CqlField::isNonValueColumn)
- .map(CqlField::name)
-
.allMatch(requiredColumns::contains);
+ .filter(CqlField::isNonValueColumn)
+ .map(CqlField::name)
+ .allMatch(requiredColumns::contains);
// Map original column position to new position in requiredSchema
positionsMap = IntStream.range(0, table.numFields())
@@ -161,7 +161,7 @@ public class SparkRowIterator extends
AbstractSparkRowIterator implements Partit
}
// Otherwise we need to only return columns requested and map to
new position in result array
- int length = noValueColumns ? cell.values.length :
cell.values.length - 1;
+ int length = !hasProjectedValueColumns ? cell.values.length :
cell.values.length - 1;
for (int index = 0; index < length; index++)
{
int position = positionsMap[index];
diff --git
a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/EndToEndTests.java
b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/EndToEndTests.java
index 4dad7c3..018f56f 100644
---
a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/EndToEndTests.java
+++
b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/EndToEndTests.java
@@ -2524,4 +2524,117 @@ public class EndToEndTests
.withColumns("Partition_Key_0", "Column_1") // PK is required
for lookup of the inserted data
.run();
}
+
+ // null values in regular columns
+
+ @ParameterizedTest
+ @MethodSource("org.apache.cassandra.bridge.VersionRunner#bridges")
+ public void testSinglePartitionKeyWithNullValueColumn(CassandraBridge
bridge)
+ {
+ Tester.builder(TestSchema.builder(bridge).withPartitionKey("a",
bridge.bigint())
+ .withColumn("c", bridge.text()))
+ .withExpectedRowCountPerSSTable(Tester.DEFAULT_NUM_ROWS)
+ .withNullRegularColumns()
+ .run();
+ }
+
+ @ParameterizedTest
+ @MethodSource("org.apache.cassandra.bridge.VersionRunner#bridges")
+ public void testMultiplePartitionKeysWithNullValueColumn(CassandraBridge
bridge)
+ {
+ Tester.builder(TestSchema.builder(bridge)
+ .withPartitionKey("a", bridge.bigint())
+ .withPartitionKey("b", bridge.text())
+ .withPartitionKey("d", bridge.aDouble())
+ .withColumn("c", bridge.text()))
+ .withExpectedRowCountPerSSTable(Tester.DEFAULT_NUM_ROWS)
+ .withNullRegularColumns()
+ .run();
+ }
+
+ @ParameterizedTest
+ @MethodSource("org.apache.cassandra.bridge.VersionRunner#bridges")
+ public void
testSinglePartitionAndClusteringKeyWithNullValueColumn(CassandraBridge bridge)
+ {
+ qt().forAll(TestUtils.cql3Type(bridge))
+ .checkAssert(clusteringKeyType ->
+
Tester.builder(TestSchema.builder(bridge).withPartitionKey("a", bridge.bigint())
+ .withClusteringKey("b",
clusteringKeyType)
+ .withColumn("c",
bridge.text()))
+
.withExpectedRowCountPerSSTable(Tester.DEFAULT_NUM_ROWS)
+ .withNullRegularColumns()
+ .run());
+ }
+
+ @ParameterizedTest
+ @MethodSource("org.apache.cassandra.bridge.VersionRunner#bridges")
+ public void testMultipleValueColumnsWithNullValueColumn(CassandraBridge
bridge)
+ {
+ qt().forAll(TestUtils.cql3Type(bridge))
+ .checkAssert(clusteringKeyType ->
+
Tester.builder(TestSchema.builder(bridge).withPartitionKey("a", bridge.bigint())
+ .withClusteringKey("b",
clusteringKeyType)
+ .withColumn("c",
bridge.text())
+ .withColumn("d",
bridge.aInt())
+ .withColumn("e",
bridge.ascii())
+ .withColumn("f",
bridge.blob()))
+
.withExpectedRowCountPerSSTable(Tester.DEFAULT_NUM_ROWS)
+ .withNullRegularColumns()
+ .run());
+ }
+
+ @ParameterizedTest
+ @MethodSource("org.apache.cassandra.bridge.VersionRunner#bridges")
+ public void testExcludeSomeColumnsWithNullValueColumn(CassandraBridge
bridge)
+ {
+ Tester.builder(TestSchema.builder(bridge).withPartitionKey("a",
bridge.bigint())
+ .withClusteringKey("b", bridge.aInt())
+ .withColumn("c", bridge.text())
+ .withColumn("d", bridge.aInt())
+ .withColumn("e", bridge.ascii())
+ .withColumn("f", bridge.blob()))
+ .withColumns("a", "b", "d")
+ .withExpectedRowCountPerSSTable(Tester.DEFAULT_NUM_ROWS)
+ .withNullRegularColumns()
+ .run();
+ }
+
+ @ParameterizedTest
+ @MethodSource("org.apache.cassandra.bridge.VersionRunner#bridges")
+ public void testStaticColumnWithNullValueColumn(CassandraBridge bridge)
+ {
+ Tester.builder(TestSchema.builder(bridge)
+ .withPartitionKey("a", bridge.uuid())
+ .withClusteringKey("b", bridge.bigint())
+ .withStaticColumn("c", bridge.aInt())
+ .withColumn("d", bridge.text()))
+ .withExpectedRowCountPerSSTable(Tester.DEFAULT_NUM_ROWS)
+ .withNullRegularColumns()
+ .run();
+ }
+
+ @ParameterizedTest
+ @MethodSource("org.apache.cassandra.bridge.VersionRunner#bridges")
+ public void testNullValueColumnWithPushDownFilter(CassandraBridge bridge)
+ {
+ int numRows = 10;
+ Tester.builder(TestSchema.builder(bridge).withPartitionKey("a",
bridge.aInt()).withColumn("b", bridge.aInt()))
+ .dontWriteRandomData()
+ .withSSTableWriter(writer -> {
+ for (int i = 0; i < numRows; i++)
+ {
+ writer.write(i, null);
+ }
+ })
+ .withFilter("a=1")
+ .withCheck((ds) -> {
+ for (Row row : ds.collectAsList())
+ {
+ int a = row.getInt(0);
+ assertEquals(1, a);
+ assertNull(row.get(1));
+ }
+ })
+ .run();
+ }
}
diff --git
a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/Tester.java
b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/Tester.java
index 43eb135..b14d744 100644
---
a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/Tester.java
+++
b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/Tester.java
@@ -91,46 +91,30 @@ public final class Tester
private final int delayBetweenSSTablesInSecs;
private final String statsClass;
private final boolean upsert;
+ private final boolean nullifyValueColumn;
- // CHECKSTYLE IGNORE: Constructor with many parameters
- private Tester(@NotNull List<CassandraVersion> versions,
- @Nullable TestSchema.Builder schemaBuilder,
- @Nullable Function<String, TestSchema.Builder>
schemaBuilderFunc,
- @NotNull List<Integer> numSSTables,
- @NotNull List<Consumer<TestSchema.TestRow>> writeListeners,
- @NotNull List<Consumer<TestSchema.TestRow>> readListeners,
- @NotNull List<Writer> writers,
- @NotNull List<Consumer<Dataset<Row>>> checks,
- @NotNull Set<String> sumFields,
- @Nullable Runnable reset,
- @Nullable String filterExpression,
- int numRandomRows, int expectedRowCount,
- boolean shouldCheckNumSSTables,
- @Nullable String[] columns,
- boolean addLastModifiedTimestamp,
- int delayBetweenSSTablesInSecs,
- @Nullable String statsClass,
- boolean upsert)
+ private Tester(Builder builder)
{
- this.versions = versions;
- this.schemaBuilder = schemaBuilder;
- this.schemaBuilderFunc = schemaBuilderFunc;
- this.numSSTables = numSSTables;
- this.writeListeners = writeListeners;
- this.readListeners = readListeners;
- this.writers = writers;
- this.checks = checks;
- this.sumFields = sumFields;
- this.reset = reset;
- this.filterExpression = filterExpression;
- this.numRandomRows = numRandomRows;
- this.expectedRowCount = expectedRowCount;
- this.shouldCheckNumSSTables = shouldCheckNumSSTables;
- this.columns = columns;
- this.addLastModifiedTimestamp = addLastModifiedTimestamp;
- this.delayBetweenSSTablesInSecs = delayBetweenSSTablesInSecs;
- this.statsClass = statsClass;
- this.upsert = upsert;
+ this.versions = builder.versions;
+ this.schemaBuilder = builder.schemaBuilder;
+ this.schemaBuilderFunc = builder.schemaBuilderFunc;
+ this.numSSTables = builder.numSSTables;
+ this.writeListeners = builder.writeListeners;
+ this.readListeners = builder.readListeners;
+ this.writers = builder.writers;
+ this.checks = builder.checks;
+ this.sumFields = builder.sumFields;
+ this.reset = builder.reset;
+ this.filterExpression = builder.filterExpression;
+ this.numRandomRows = builder.numRandomRows;
+ this.expectedRowCount = builder.expectedRowCount;
+ this.shouldCheckNumSSTables = builder.shouldCheckNumSSTables;
+ this.columns = builder.columns;
+ this.addLastModifiedTimestamp = builder.addLastModifiedTimestamp;
+ this.delayBetweenSSTablesInSecs = builder.delayBetweenSSTablesInSecs;
+ this.statsClass = builder.statsClass;
+ this.upsert = builder.upsert;
+ this.nullifyValueColumn = builder.nullRegularColumns;
}
static Builder builder(@NotNull TestSchema.Builder schemaBuilder)
@@ -194,6 +178,7 @@ public final class Tester
private int delayBetweenSSTablesInSecs = 0;
private String statsClass = null;
private boolean upsert = false;
+ private boolean nullRegularColumns = false;
private Builder(@NotNull TestSchema.Builder schemaBuilder)
{
@@ -337,28 +322,16 @@ public final class Tester
return this;
}
+ public Builder withNullRegularColumns()
+ {
+ this.nullRegularColumns = true;
+ return this;
+ }
+
void run()
{
Preconditions.checkArgument(schemaBuilder != null ||
schemaBuilderFunc != null);
- new Tester(versions,
- schemaBuilder,
- schemaBuilderFunc,
- numSSTables,
- writeListeners,
- readListeners,
- writers,
- checks,
- sumFields,
- reset,
- filterExpression,
- numRandomRows,
- expectedRowCount,
- shouldCheckNumSSTables,
- columns,
- addLastModifiedTimestamp,
- delayBetweenSSTablesInSecs,
- statsClass,
- upsert).run();
+ new Tester(this).run();
}
}
@@ -392,34 +365,34 @@ public final class Tester
.collect(Collectors.toMap(Function.identity(),
ignore -> new MutableLong()));
Map<String, TestSchema.TestRow> rows = new
HashMap<>(numRandomRows);
- IntStream.range(0, numSSTables).forEach(ssTable ->
- schema.writeSSTable(directory, bridge, partitioner,
upsert, writer ->
- IntStream.range(0, numRandomRows).forEach(row -> {
- TestSchema.TestRow testRow;
- do
- {
- testRow = schema.randomRow();
- }
- while (rows.containsKey(testRow.getKey()));
// Don't write duplicate rows
-
- for (Consumer<TestSchema.TestRow>
writeListener : writeListeners)
- {
- writeListener.accept(testRow);
- }
-
- for (String sumField : sumFields)
- {
- sum.get(sumField).add((Number)
testRow.get(sumField));
- }
- rows.put(testRow.getKey(), testRow);
-
- Object[] values = testRow.allValues();
- if (upsert)
- {
- rotate(values,
schema.partitionKeys.size() + schema.clusteringKeys.size());
- }
- writer.write(values);
- })));
+ IntStream.range(0, numSSTables).forEach(ssTable ->
schema.writeSSTable(directory, bridge, partitioner, upsert, writer -> {
+ IntStream.range(0, numRandomRows).forEach(row -> {
+ TestSchema.TestRow testRow;
+ do
+ {
+ testRow = schema.randomRow(nullifyValueColumn);
+ }
+ while (rows.containsKey(testRow.getKey())); // Don't
write duplicate rows
+
+ for (Consumer<TestSchema.TestRow> writeListener :
writeListeners)
+ {
+ writeListener.accept(testRow);
+ }
+
+ for (String sumField : sumFields)
+ {
+ sum.get(sumField).add((Number) testRow.get(sumField));
+ }
+ rows.put(testRow.getKey(), testRow);
+
+ Object[] values = testRow.allValues();
+ if (upsert)
+ {
+ rotate(values, schema.partitionKeys.size() +
schema.clusteringKeys.size());
+ }
+ writer.write(values);
+ });
+ }));
int sstableCount = numSSTables;
// Write any custom SSTables e.g. overwriting existing data or
tombstones
diff --git
a/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/EmptyStreamScanner.java
b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/EmptyStreamScanner.java
index 822a8cf..fcd3712 100644
---
a/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/EmptyStreamScanner.java
+++
b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/EmptyStreamScanner.java
@@ -40,6 +40,12 @@ public class EmptyStreamScanner implements StreamScanner<Rid>
{
}
+ @Override
+ public boolean hasMoreColumns()
+ {
+ return false;
+ }
+
@Override
public void close()
{
diff --git
a/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/StreamScanner.java
b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/StreamScanner.java
index 2a4e975..cdd0c72 100644
---
a/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/StreamScanner.java
+++
b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/StreamScanner.java
@@ -77,4 +77,9 @@ public interface StreamScanner<Type> extends Closeable
* @throws IOException
*/
void advanceToNextColumn() throws IOException;
+
+ /**
+ * @return {@code true} if the scanner has more columns to consume, {@code
false} otherwise
+ */
+ boolean hasMoreColumns();
}
diff --git
a/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/common/IndexIterator.java
b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/common/IndexIterator.java
index 1103153..5228c17 100644
---
a/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/common/IndexIterator.java
+++
b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/reader/common/IndexIterator.java
@@ -137,6 +137,12 @@ public class IndexIterator<ReaderType extends
IIndexReader> implements StreamSca
return finished.get() == readers.size();
}
+ @Override
+ public boolean hasMoreColumns()
+ {
+ return true;
+ }
+
public void advanceToNextColumn()
{
if (closed.get())
diff --git
a/cassandra-bridge/src/main/java/org/apache/cassandra/spark/utils/test/TestSchema.java
b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/utils/test/TestSchema.java
index fb33da4..2e6b0a8 100644
---
a/cassandra-bridge/src/main/java/org/apache/cassandra/spark/utils/test/TestSchema.java
+++
b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/utils/test/TestSchema.java
@@ -35,6 +35,7 @@ import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.function.Consumer;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
@@ -500,16 +501,33 @@ public final class TestSchema
public TestRow randomRow()
{
- Object[] values = new Object[allFields.size()];
- for (CqlField field : allFields)
+ return randomRow(false);
+ }
+
+ public TestRow randomRow(boolean nullifyValueColumn)
+ {
+ return randomRow(field -> nullifyValueColumn && field.isValueColumn());
+ }
+
+ private TestRow randomRow(Predicate<CqlField> nullifiedFields)
+ {
+ final Object[] values = new Object[allFields.size()];
+ for (final CqlField field : allFields)
{
- if (field.type().getClass().getSimpleName().equals("Blob") &&
blobSize != null)
+ if (nullifiedFields.test(field))
{
- values[field.position()] =
RandomUtils.randomByteBuffer(blobSize);
+ values[field.position()] = null;
}
else
{
- values[field.position()] =
field.type().randomValue(minCollectionSize);
+ if (field.type().getClass().getSimpleName().equals("Blob") &&
blobSize != null)
+ {
+ values[field.position()] =
RandomUtils.randomByteBuffer(blobSize);
+ }
+ else
+ {
+ values[field.position()] =
field.type().randomValue(minCollectionSize);
+ }
}
}
return new TestRow(values);
diff --git
a/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java
b/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java
index 3ebe865..460ced0 100644
---
a/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java
+++
b/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java
@@ -125,6 +125,12 @@ public abstract class AbstractStreamScanner implements
StreamScanner<Rid>, Close
columnData.consume();
}
+ @Override
+ public boolean hasMoreColumns()
+ {
+ return columns != null && columns.hasNext();
+ }
+
// CHECKSTYLE IGNORE: Long method
@Override
public boolean hasNext() throws IOException
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org
For additional commands, e-mail: commits-h...@cassandra.apache.org
