skoppu22 commented on code in PR #109:
URL:
https://github.com/apache/cassandra-analytics/pull/109#discussion_r2081780842
##########
cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteUdtTest.java:
##########
@@ -32,7 +32,7 @@
import org.apache.spark.sql.SparkSession;
import org.jetbrains.annotations.NotNull;
-import static org.apache.cassandra.testing.TestUtils.DC1_RF3;
+import static org.apache.cassandra.testing.TestUtils.DC1_RF1;
Review Comment:
Done
##########
cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkWriteUdtTest.java:
##########
@@ -88,17 +141,298 @@ void testWriteWithNestedUdt()
assertThat(result.hasNext()).isTrue();
validateWritesWithDriverResultSet(df.collectAsList(),
queryAllDataWithDriver(NESTED_TABLE_NAME),
-
BulkWriteUdtTest::defaultRowFormatter);
+ BulkWriteUdtTest::udtRowFormatter);
+ }
+
+ @Test
+ void testListOfUdts()
+ {
+ int numRowsInserted = populateListOfUdts();
+
+ // Create a spark frame with the data inserted during the setup
+ Dataset<Row> sourceData =
bulkReaderDataFrame(LIST_OF_UDT_SOURCE_TABLE).load();
+ assertThat(sourceData.count()).isEqualTo(numRowsInserted);
+
+ // Insert the dataset containing list of UDTs, and UDT itself has
collections in it
+ bulkWriterDataFrameWriter(sourceData, LIST_OF_UDT_DEST_TABLE).save();
+ validateWritesWithDriverResultSet(sourceData.collectAsList(),
+ queryAllDataWithDriver(LIST_OF_UDT_DEST_TABLE),
+ BulkWriteUdtTest::listOfUdtRowFormatter);
+ }
+
+ private int populateListOfUdts()
+ {
+ // table(id, list<udt(list<>, set<>, map<>)>)
+ // insert list of UDTs, and each UDT has a list, set and map
+ String insertIntoListOfUdts = "INSERT INTO %s (id, udtlist) VALUES
(%d, [{f1:['value %d'], f2:{%d}, f3:{%d : 'value %d'}}])";
+
+ int i = 0;
+ for (; i < ROW_COUNT; i++)
+ {
+
cluster.schemaChangeIgnoringStoppedInstances(String.format(insertIntoListOfUdts,
LIST_OF_UDT_SOURCE_TABLE, i, i, i, i, i));
+ }
+
+ // test null cases
+ cluster.schemaChangeIgnoringStoppedInstances(String.format("insert
into %s (id) values (%d)",
+ LIST_OF_UDT_SOURCE_TABLE, i++));
+ cluster.schemaChangeIgnoringStoppedInstances(String.format("insert
into %s (id, udtlist) values (%d, null)",
+ LIST_OF_UDT_SOURCE_TABLE, i++));
+ cluster.schemaChangeIgnoringStoppedInstances(String.format("insert
into %s (id, udtlist) values (%d, [{f1:null, f2:null, f3:null}])",
+ LIST_OF_UDT_SOURCE_TABLE, i++));
+
+ return i;
+ }
+
+ @Test
+ void testSetOfUdts()
+ {
+ int numRowsInserted = populateSetOfUdts();
+ // Create a spark frame with the data inserted during the setup
+ Dataset<Row> sourceData =
bulkReaderDataFrame(SET_OF_UDT_SOURCE_TABLE).load();
+ assertThat(sourceData.count()).isEqualTo(numRowsInserted);
+
+ // Insert the dataset containing set of UDTs, and UDT itself has
collections in it
+ bulkWriterDataFrameWriter(sourceData, SET_OF_UDT_DEST_TABLE).save();
+ validateWritesWithDriverResultSet(sourceData.collectAsList(),
+ queryAllDataWithDriver(SET_OF_UDT_DEST_TABLE),
+ BulkWriteUdtTest::setOfUdtRowFormatter);
+ }
+
+ private int populateSetOfUdts()
+ {
+ // table(id, set<udt(list<>, set<>, map<>)>)
+ // insert set of UDTs, and UDT has a list, set and map inside it
+ String insertIntoSetOfUdts = "INSERT INTO %s (id, udtset) VALUES (%d,
" +
+ "{{f1:['value %d'], f2:{%d}, f3:{%d : 'value %d'}}})";
+
+ int i = 0;
+ for (; i < ROW_COUNT; i++)
+ {
+
cluster.schemaChangeIgnoringStoppedInstances(String.format(insertIntoSetOfUdts,
SET_OF_UDT_SOURCE_TABLE,
+ i, i, i, i, i));
+ }
+
+ // test null cases
+ cluster.schemaChangeIgnoringStoppedInstances(String.format("insert
into %s (id) values (%d)",
+ SET_OF_UDT_SOURCE_TABLE, i++));
+ cluster.schemaChangeIgnoringStoppedInstances(String.format("insert
into %s (id, udtset) values (%d, null)",
+ SET_OF_UDT_SOURCE_TABLE, i++));
+ cluster.schemaChangeIgnoringStoppedInstances(String.format("insert
into %s (id, udtset) values (%d, " +
+ "{{f1:null, f2:null, f3:null}})",
+ SET_OF_UDT_SOURCE_TABLE, i++));
+
Review Comment:
Changed to use execute()
--
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: commits-unsubscr...@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org
For additional commands, e-mail: commits-h...@cassandra.apache.org
