jyothsnakonisa commented on code in PR #220:
URL:
https://github.com/apache/cassandra-analytics/pull/220#discussion_r3469331707
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TableSchema.java:
##########
@@ -316,6 +349,45 @@ static void validateNoSecondaryIndexes(TableInfoProvider
tableInfo)
}
}
+ public Set<String> getIndexStatements()
+ {
+ return indexStatements;
+ }
+
+ @VisibleForTesting
+ static boolean isCassandra5OrLater(String version)
Review Comment:
Can u check the correctness of this method for a version like 3.11.x? May be
worth adding tests for some of the methods you added here in this class
##########
cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/common/SSTables.java:
##########
@@ -22,14 +22,52 @@
import java.nio.file.Path;
import org.apache.cassandra.bridge.SSTableDescriptor;
+import org.apache.cassandra.spark.data.FileType;
public final class SSTables
{
+ /**
+ * Suffix identifying the primary SSTable data component, e.g. "-Data.db".
+ * The leading '-' is significant: it excludes SAI per-index components
such as
+ * "...+TermsData.db" that also end with "Data.db".
+ */
+ private static final String DATA_COMPONENT_SUFFIX = "-" +
FileType.DATA.getFileSuffix();
+
+ /**
+ * Glob matching primary SSTable data components, e.g. "*-Data.db".
+ * Suitable for {@link java.nio.file.Files#newDirectoryStream(Path,
String)}.
+ */
+ public static final String DATA_COMPONENT_GLOB = "*" +
DATA_COMPONENT_SUFFIX;
+
private SSTables()
{
throw new IllegalStateException(getClass() + " is static utility class
and shall not be instantiated");
}
+ /**
+ * Determine whether the given file name is a primary SSTable data
component ("<descriptor>-Data.db").
+ * The leading '-' check excludes SAI per-index components such as
"...+TermsData.db" which also end with "Data.db".
+ *
+ * @param fileName file name (not a full path)
+ * @return true if the name is a primary data component
+ */
+ public static boolean isDataComponent(String fileName)
+ {
+ return fileName.endsWith(DATA_COMPONENT_SUFFIX);
+ }
+
+ /**
+ * Determine whether the given path is a primary SSTable data component
("<descriptor>-Data.db").
+ *
+ * @param path file path
+ * @return true if the path's file name is a primary data component
+ * @see #isDataComponent(String)
+ */
+ public static boolean isDataComponent(Path path)
+ {
+ return isDataComponent(path.getFileName().toString());
Review Comment:
Should we do a null check?
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TableSchema.java:
##########
@@ -308,6 +300,47 @@ private static void
validateDataframeFieldsInTable(TableInfoProvider tableInfo,
Preconditions.checkArgument(unknownFields.isEmpty(), "Unknown fields
in data frame => " + unknownFields);
}
+ /**
+ * Validates secondary index constraints for bulk write operations.
+ * <p>
+ * When the cluster is Cassandra 5.0+ and ALL indexes are SAI, the write
is allowed because
+ * SAI index components are generated alongside SSTables and are
immediately queryable after import.
+ * <p>
+ * When any index is non-SAI (legacy 2i), the write is blocked unless
SKIP_SECONDARY_INDEX_CHECK is set.
+ *
+ * @param tableInfo the table info provider
+ * @param skipSecondaryIndexCheck whether the user explicitly opted out
of the check
+ * @param indexStatements the CREATE INDEX statements for the
table
+ * @param lowestCassandraVersion the lowest Cassandra version in the
cluster
+ */
+ static void validateSecondaryIndexes(TableInfoProvider tableInfo,
Review Comment:
Can u add tests for this method?
##########
cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/common/SSTables.java:
##########
Review Comment:
Could you please add test for new methods that are being added specially for
SAI file names?
##########
cassandra-five-zero-types/src/main/java/org/apache/cassandra/spark/reader/StorageAttachedIndexApplier.java:
##########
Review Comment:
Since this is making important change to the schema, could you please add
tests for this class?
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TableSchema.java:
##########
@@ -308,6 +300,47 @@ private static void
validateDataframeFieldsInTable(TableInfoProvider tableInfo,
Preconditions.checkArgument(unknownFields.isEmpty(), "Unknown fields
in data frame => " + unknownFields);
}
+ /**
+ * Validates secondary index constraints for bulk write operations.
+ * <p>
+ * When the cluster is Cassandra 5.0+ and ALL indexes are SAI, the write
is allowed because
+ * SAI index components are generated alongside SSTables and are
immediately queryable after import.
+ * <p>
+ * When any index is non-SAI (legacy 2i), the write is blocked unless
SKIP_SECONDARY_INDEX_CHECK is set.
+ *
+ * @param tableInfo the table info provider
+ * @param skipSecondaryIndexCheck whether the user explicitly opted out
of the check
+ * @param indexStatements the CREATE INDEX statements for the
table
+ * @param lowestCassandraVersion the lowest Cassandra version in the
cluster
+ */
+ static void validateSecondaryIndexes(TableInfoProvider tableInfo,
Review Comment:
Is `TableInfoProvider tableInfo,` parameter needed? can we may be do
`if (indexStatements.isEmpty()) ` instead of `if
(!tableInfo.hasSecondaryIndex())`
##########
cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java:
##########
@@ -222,6 +224,20 @@ private static Pair<KeyspaceMetadata, TableMetadata>
updateSchema(Schema schema,
}
tableMetadata.columns().forEach(columnValidator);
+
+ // This builder always produces an index-less table (indexes are
applied later by the 5.0 bridge), so a
+ // rebuild never carries indexes even if the caller passed index
statements. buildSchema runs repeatedly per
+ // table in a JVM; if an earlier call already registered indexes, copy
them onto this rebuild so it matches
+ // the registered table.
+ if (maybeExistingTableMetadata != null
+ && !maybeExistingTableMetadata.indexes.isEmpty()
+ && tableMetadata.indexes.isEmpty())
+ {
+ tableMetadata = tableMetadata.unbuild()
+
.indexes(maybeExistingTableMetadata.indexes)
+ .build();
+ }
+
Review Comment:
RecordWriter is using 5 arg buildSchema() method which is not attaching
index statements, you could eliminate copying indexes here if you use the 8 arg
buildSchema() method in RecordWriter and can eventually get rid of 5 arg
buildSchema() method
https://github.com/apache/cassandra-analytics/blob/0eb3a7f85dbef0500e947f4a8d9a1ae3682bdc60/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RecordWriter.java#L119
```
cqlTable = writerContext.bridge()
.buildSchema(writerContext.schema().getTableSchema().createStatement,
writerContext.job().qualifiedTableName().keyspace(),
IGNORED_REPLICATION_FACTOR,
writerContext.cluster().getPartitioner(),
writerContext.schema().getUserDefinedTypeStatements(),
null,
writerContext.schema().getTableSchema().getIndexStatements(),
false);
```
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TableSchema.java:
##########
@@ -308,6 +300,47 @@ private static void
validateDataframeFieldsInTable(TableInfoProvider tableInfo,
Preconditions.checkArgument(unknownFields.isEmpty(), "Unknown fields
in data frame => " + unknownFields);
}
+ /**
+ * Validates secondary index constraints for bulk write operations.
+ * <p>
+ * When the cluster is Cassandra 5.0+ and ALL indexes are SAI, the write
is allowed because
+ * SAI index components are generated alongside SSTables and are
immediately queryable after import.
+ * <p>
+ * When any index is non-SAI (legacy 2i), the write is blocked unless
SKIP_SECONDARY_INDEX_CHECK is set.
+ *
+ * @param tableInfo the table info provider
+ * @param skipSecondaryIndexCheck whether the user explicitly opted out
of the check
+ * @param indexStatements the CREATE INDEX statements for the
table
+ * @param lowestCassandraVersion the lowest Cassandra version in the
cluster
+ */
+ static void validateSecondaryIndexes(TableInfoProvider tableInfo,
+ boolean skipSecondaryIndexCheck,
+ Set<String> indexStatements,
+ String lowestCassandraVersion)
+ {
+ if (!tableInfo.hasSecondaryIndex())
+ {
+ return; // No indexes — nothing to validate
+ }
+
+ if (isSaiWrite(indexStatements, lowestCassandraVersion))
+ {
+ LOGGER.info("Table has SAI indexes on Cassandra 5.0+. SAI index
components will be generated "
+ + "alongside SSTables. indexCount={}",
indexStatements.size());
+ return;
+ }
+
+ if (skipSecondaryIndexCheck)
+ {
+ LOGGER.warn("Bulk writing to tables with SecondaryIndexes will
have an asynchronous index rebuild "
+ + "take place automatically after writing. Reads against
the index during this time "
+ + "window will produce inconsistent or stale results
until index rebuild is complete.");
+ return;
+ }
+
+ throw new UnsupportedAnalyticsOperationException("Bulkwriter doesn't
support secondary indexes");
Review Comment:
How about changing exception message to ""Bulkwriter doesn't support non-SAI
indexes."
##########
cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridgeImplementation.java:
##########
@@ -281,10 +282,17 @@ public CqlTable buildSchema(String createStatement,
Partitioner partitioner,
Set<String> udts,
@Nullable UUID tableId,
- int indexCount,
+ Set<String> indexStatements,
boolean enableCdc)
{
- return new SchemaBuilder(createStatement, keyspace, replicationFactor,
partitioner, cassandraTypes -> udts, tableId, indexCount, enableCdc).build();
+ CqlTable cqlTable = new SchemaBuilder(createStatement, keyspace,
replicationFactor, partitioner,
+ cassandraTypes -> udts, tableId,
indexStatements, enableCdc).build();
+ // SAI is a Cassandra 5.0+ feature: register any Storage Attached
Index definitions on the table metadata
+ // after the shared SchemaBuilder has built/registered the
(index-less) table.
+ // buildSchema is invoked repeatedly within a JVM (the index-carrying
per-partition RecordWriter included),
+ // so this re-applies the SAI definitions whenever a rebuild left the
table without them.
+ StorageAttachedIndexApplier.maybeApply(keyspace, cqlTable.table(),
indexStatements);
+ return cqlTable;
Review Comment:
Schema Is created first and then SAI indexes are attached in the next step,
I wondering about the brief moment between two statements where the schema
doesn't have SAI indexes.
WDYT about merging both operations into a single CassandraSchema.apply in
the 5.0 bridge? you might have to override CassandraSchema.apply call entirely
from the 5.0 bridge and don't use SchemaBuilder's constructor at all for the
schema update.
I am open to other approaches too
##########
cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java:
##########
@@ -59,6 +60,9 @@ public final class CqlUtils
private static final Pattern ESCAPED_DOUBLE_BACKSLASH =
Pattern.compile("\\\\");
private static final Pattern COMPACTION_STRATEGY_PATTERN =
Pattern.compile("compaction\\s*=\\s*\\{\\s*'class'\\s*:\\s*'([^']+)'");
+ private static final Pattern MULTI_WHITESPACE_PATTERN =
Pattern.compile("\\s+");
+ private static final Pattern SAI_USING_PATTERN = Pattern.compile("USING
'[^']*STORAGEATTACHEDINDEX'");
Review Comment:
This pattern matches with something like "USING
'PrefixStorageAttachedIndex'" which is incorrect, how about using something
like "Pattern.compile("USING '([^']*\\.)?STORAGEATTACHEDINDEX'")"
--
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]
