yifan-c commented on code in PR #19:
URL:
https://github.com/apache/cassandra-analytics/pull/19#discussion_r1413883025
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/clients/SidecarInstanceImpl.java:
##########
@@ -90,14 +90,14 @@ public String toString()
private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException
{
- LOGGER.warn("Falling back to JDK deserialization");
+ LOGGER.debug("Falling back to JDK deserialization");
Review Comment:
+1 on changing the log level to debug. If warning on serialization method is
desired, it should be logged separated and warn just once.
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java:
##########
@@ -45,5 +45,10 @@ public enum WriterOptions implements WriterOption
ROW_BUFFER_MODE,
SSTABLE_DATA_SIZE_IN_MB,
TTL,
- TIMESTAMP
+ TIMESTAMP,
+ /**
+ * Option that specifies whether the identifiers (i.e. keyspace, table
name, column names) should be quoted to
+ * support mixed case and reserved keyword names for these fields.
+ */
+ QUOTE_IDENTIFIERS,
Review Comment:
why it is a writer option?
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java:
##########
@@ -343,7 +346,7 @@ private CompletionStage<Map<String, AvailabilityHint>>
createSnapshot(RingRespon
snapshotName, keyspace, table, datacenter,
ringEntry.fqdn());
Review Comment:
The log message should use `maybeQuotedKeyspace` and `maybeQuotedTable`
instead.
The parameters in the log message above (`LOGGER.warn("Skip snapshot
creating when node is joining or down "`) should be updated too.
And the log messages in `listInstance` method.
##########
cassandra-analytics-core/src/main/spark3/org/apache/cassandra/spark/sparksql/CassandraScanBuilder.java:
##########
@@ -135,17 +136,16 @@ 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
+
+ if (partitionKeyValues.isEmpty())
{
- return new ArrayList<>();
+ return Collections.emptyList();
}
+
+ List<List<String>> orderedValues =
partitionKeyColumnNames.stream().map(partitionKeyValues::get).collect(Collectors.toList());
+ return FilterUtils.cartesianProduct(orderedValues).stream()
+ .map(this::buildFilter)
+ .collect(Collectors.toList());
Review Comment:
It looks like only the if and else blocks are switched. I would discourage
such refactoring.
We can keep this change since it is done, but let's refrain from doing so in
the future patches.
##########
cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/EndToEndTests.java:
##########
@@ -1851,19 +1848,19 @@ public void testUdtsWithNulls(CassandraBridge bridge)
for (long pk = 0; pk < Tester.DEFAULT_NUM_ROWS; pk++)
{
Map<String, Object> value = ImmutableMap.of(
- pk < midPoint ? "a" : "b",
RandomUtils.randomValue(bridge.text()).toString(),
- "c",
RandomUtils.randomValue(bridge.text()).toString());
+ pk < midPoint ? "a" : "b",
RandomUtils.randomValue(bridge.text()).toString(),
+ "c", RandomUtils.randomValue(bridge.text()).toString());
Review Comment:
It is hard to read. It is not obvious that those 2 lines are the parameters
for the `of` method without indentations.
Please restore the formatting.
##########
cassandra-bridge/src/main/java/org/apache/cassandra/spark/data/CqlField.java:
##########
@@ -263,7 +263,7 @@ public CqlField(boolean isPartitionKey,
this.isPartitionKey = isPartitionKey;
this.isClusteringColumn = isClusteringColumn;
this.isStaticColumn = isStaticColumn;
- this.name = name.replaceAll("\"", "");
+ this.name = name;
Review Comment:
> we no longer need to drop the quotes, since we no longer pass the quoted
string from here:
https://github.com/apache/cassandra-analytics/pull/19/files#diff-143d0bc336629bfe78bf9f0c64d83803701d38b22a4be736323277862b784b25R522
This comment is confusing to me. The new test cases in `EndToEndTests` pass
quoted names, and the tests fail if quotation marks are removed.
For UDT, the names are always unquoted, and there is `fieldsString` to quote
back if needed. But it is not true for plain `CqlField`. There is some
inconsistency.
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java:
##########
@@ -343,7 +346,7 @@ private CompletionStage<Map<String, AvailabilityHint>>
createSnapshot(RingRespon
snapshotName, keyspace, table, datacenter,
ringEntry.fqdn());
SidecarInstance sidecarInstance = new
SidecarInstanceImpl(ringEntry.fqdn(), sidecarClientConfig.effectivePort());
createSnapshotFuture = sidecar
- .createSnapshot(sidecarInstance,
keyspace, table, snapshotName)
+ .createSnapshot(sidecarInstance,
maybeQuotedKeyspace, maybeQuotedTable, snapshotName)
Review Comment:
What if `maybeQuotedKeyspace` and/or `maybeQuotedTable` are null in this
method. You added the check in the shutdown hook, should it be checked here?
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataSourceHelper.java:
##########
@@ -90,7 +94,7 @@ public static DataLayer getDataLayer(
}
}
- protected static Cache<Map<String, String>, CassandraDataLayer>
getCassandraDataLayerCache()
+ public static Cache<Map<String, String>, CassandraDataLayer>
getCassandraDataLayerCache()
Review Comment:
Why are the methods (including others in the file) are changed from
`protected` to `public`?
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java:
##########
@@ -888,14 +911,20 @@ protected void dialHome(@NotNull ClientConfig options)
protected void clearSnapshot(Set<? extends SidecarInstance> clusterConfig,
@NotNull ClientConfig options)
{
+ if (maybeQuotedKeyspace == null || maybeQuotedTable == null)
+ {
+ LOGGER.info("Bridge was never initialized. Skipping clearing
snapshots. This implies snapshots were never created");
Review Comment:
Should it be a warn?
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java:
##########
@@ -128,6 +128,7 @@ public class BulkSparkConf implements Serializable
public final int commitThreadsPerInstance;
protected final int effectiveSidecarPort;
protected final int userProvidedSidecarPort;
+ public boolean quoteIdentifiers;
Review Comment:
Can it be `final`? It does not seem to be updated.
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java:
##########
@@ -56,21 +56,21 @@ public class BulkSparkConf implements Serializable
private static final Logger LOGGER =
LoggerFactory.getLogger(BulkSparkConf.class);
public static final String 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";
+ + " --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";
Review Comment:
In principle, we should not change the code only for the sake of correcting
the style/format. The noise it creates voids the little benefit provided.
##########
cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/EndToEndTests.java:
##########
@@ -154,15 +154,15 @@ public void testBasicSingleClusteringKey(CassandraBridge
bridge)
public void testSingleClusteringKeyOrderBy(CassandraBridge bridge)
{
qt().forAll(TestUtils.cql3Type(bridge), TestUtils.sortOrder())
- .checkAssert((clusteringKeyType, sortOrder) ->
+ .checkAssert((clusteringKeyType, sortOrder) -> {
Tester.builder(TestSchema.builder()
.withPartitionKey("a",
bridge.bigint())
.withClusteringKey("b",
clusteringKeyType)
.withColumn("c", bridge.bigint())
.withSortOrder(sortOrder))
.withExpectedRowCountPerSSTable(Tester.DEFAULT_NUM_ROWS)
- .run()
- );
+ .run();
+ });
Review Comment:
unnecessary refactoring and many others in this file.
##########
cassandra-bridge/src/main/java/org/apache/cassandra/spark/data/DataLayer.java:
##########
@@ -171,8 +171,8 @@ public CassandraVersion version()
public abstract boolean isInPartition(int partitionId, BigInteger token,
ByteBuffer key);
public List<PartitionKeyFilter> partitionKeyFiltersInRange(
- int partitionId,
- List<PartitionKeyFilter> partitionKeyFilters) throws
NoMatchFoundException
+ int partitionId,
+ List<PartitionKeyFilter> partitionKeyFilters) throws NoMatchFoundException
Review Comment:
It is still hard to read after the change. Please restore the format. The
change is unrelated and unnecessary.
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java:
##########
@@ -239,12 +241,23 @@ public void initialize(@NotNull ClientConfig options)
LOGGER.info("Initialized Cassandra Bulk Reader with
effectiveNumberOfCores={}", effectiveNumberOfCores);
}
- private int initBulkReader(@NotNull ClientConfig options,
- CompletableFuture<NodeSettings>
nodeSettingsFuture,
- CompletableFuture<RingResponse> ringFuture)
throws ExecutionException, InterruptedException
+ private int initBulkReader(@NotNull ClientConfig options) throws
ExecutionException, InterruptedException
{
Preconditions.checkArgument(keyspace != null, "Keyspace must be
non-null for Cassandra Bulk Reader");
Preconditions.checkArgument(table != null, "Table must be non-null for
Cassandra Bulk Reader");
+
ShutdownHookManager.addShutdownHook(org.apache.spark.util.ShutdownHookManager.TEMP_DIR_SHUTDOWN_PRIORITY(),
+ ScalaFunctions.wrapLambda(() ->
shutdownHook(options)));
+
+ NodeSettings nodeSettings = sidecar.nodeSettings().get();
+ String cassandraVersion =
getEffectiveCassandraVersionForRead(clusterConfig, nodeSettings);
+ Partitioner partitioner = Partitioner.from(nodeSettings.partitioner());
+ bridge = CassandraBridgeFactory.get(cassandraVersion);
+ // optionally quote identifiers if the option has been set, we need an
instance for the bridge
+ maybeQuoteKeyspaceAndTable();
Review Comment:
`maybeQuoteKeyspaceAndTable` was already called in the constructor. I guess
it is preferred to call it here. But it does not harm with running the method
twice.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
