Merge pull request #4 from databendcloud/fix/sync-json-schemaless

fix: json schemaless data sync failed

Author: hantmac

src/main/java/com/databend/kafka/connect/sink/DatabendWriter.java

@@ -55,7 +55,6 @@ protected void onConnect(final Connection connection) throws SQLException {
     public void writeSchemaLessData(final Collection<Record> records) throws SQLException, TableAlterOrCreateException {
         final com.databend.jdbc.DatabendConnection connection = (com.databend.jdbc.DatabendConnection) cachedConnectionProvider.getConnection();
         log.info("DatabendWriter Writing {} records", records.size());
-        // new ObjectMapper
         ObjectMapper objectMapper = new ObjectMapper();
 
         StringBuilder sb = new StringBuilder();
@@ -65,7 +64,7 @@ public void writeSchemaLessData(final Collection<Record> records) throws SQLExce
                 tableId = destinationTable(record.getTopic());
                 Map<String, Data> recordMap = record.getJsonMap();
 
-                // 创建一个新的 Map 来存储转换后的数据
+                // create a new map to store the transformed data
                 Map<String, Object> transformedMap = new HashMap<>();
 
                 for (Map.Entry<String, Data> entry : recordMap.entrySet()) {
@@ -79,7 +78,6 @@ public void writeSchemaLessData(final Collection<Record> records) throws SQLExce
                         case INT16:
                         case INT32:
                         case INT64:
-                            log.info("DatabendWriter Writing record int data");
                             value = Integer.parseInt(data.getObject().toString());
                             break;
                         case FLOAT32:
@@ -90,11 +88,9 @@ public void writeSchemaLessData(final Collection<Record> records) throws SQLExce
                             value = Boolean.parseBoolean(data.getObject().toString());
                             break;
                         case STRING:
-                            log.info("DatabendWriter Writing record string data");
                             value = data.getObject().toString();
                             break;
                         default:
-                            log.info("DatabendWriter Writing record string data");
                             value = data.getObject().toString();
                             break;
                     }
@@ -107,34 +103,59 @@ public void writeSchemaLessData(final Collection<Record> records) throws SQLExce
                 String json = objectMapper.writeValueAsString(transformedMap);
                 sb.append(json).append("\n");
             }
+
+            // if there are no records to write, return
+            if (sb.length() == 0) {
+                log.info("No records to write");
+                return;
+            }
+
             String jsonStr = sb.toString();
-//            log.info("DatabendWriter Writing jsonStr is: {}", jsonStr);
+
+            // create uuid for the stage path
             String uuid = UUID.randomUUID().toString();
-            String stagePrefix = String.format("%s/%s/%s/%s/%s/%s/%s/",
+            String stagePath = String.format("%s/%s/%s/%s/%s/%s/%s",
                     LocalDateTime.now().getYear(),
                     LocalDateTime.now().getMonthValue(),
                     LocalDateTime.now().getDayOfMonth(),
                     LocalDateTime.now().getHour(),
                     LocalDateTime.now().getMinute(),
                     LocalDateTime.now().getSecond(),
                     uuid);
-            InputStream inputStream = new ByteArrayInputStream(jsonStr.getBytes(StandardCharsets.UTF_8));
-            String fileName = String.format("%s.%s", uuid, "ndjson");
-            connection.uploadStream("~", stagePrefix, inputStream, fileName, jsonStr.length(), false);
+
+            String fileName = String.format("%s.ndjson", uuid);
+
+            byte[] jsonBytes = jsonStr.getBytes(StandardCharsets.UTF_8);
+            InputStream inputStream = new ByteArrayInputStream(jsonBytes);
+            int contentLength = jsonBytes.length;
+
+            log.info("Uploading data, file size: {} bytes", contentLength);
+
+            connection.uploadStream("~", stagePath, inputStream, fileName, contentLength, false);
+
             assert tableId != null;
+
+            String stagePlusFileName = String.format("@~/%s/%s", stagePath, fileName);
+
+            log.info("Copying data from stage: {}", stagePlusFileName);
+
             String copyIntoSQL = String.format(
                     "COPY INTO %s FROM %s FILE_FORMAT = (type = NDJSON missing_field_as = FIELD_DEFAULT COMPRESSION = AUTO) " +
                             "PURGE = %b FORCE = %b DISABLE_VARIANT_CHECK = %b",
                     tableId,
-                    String.format("@~/%s/%s", stagePrefix, fileName),
+                    stagePlusFileName,
                     true,
                     true,
                     true
             );
+
             try {
+                log.info("Executing COPY INTO SQL: {}", copyIntoSQL);
                 connection.createStatement().execute(copyIntoSQL);
+                log.info("COPY INTO completed successfully");
             } catch (Exception e) {
                 log.error("DatabendWriter writeSchemaLessData error: {}", e);
+                throw e; // throw the exception to the caller
             }
         } catch (TableAlterOrCreateException e) {
             throw e;

target/components/packages/databendCloud-databend-kafka-connect-0.0.1-SNAPSHOT/databendCloud-databend-kafka-connect-0.0.1-SNAPSHOT/manifest.json

@@ -31,5 +31,5 @@
     "url" : "https://www.apache.org/licenses/LICENSE-2.0"
   } ],
   "component_types" : [ "sink" ],
-  "release_date" : "2024-08-06"
+  "release_date" : "2025-03-18"
 }

target/manifest.json

@@ -31,5 +31,5 @@
     "url" : "https://www.apache.org/licenses/LICENSE-2.0"
   } ],
   "component_types" : [ "sink" ],
-  "release_date" : "2024-08-06"
+  "release_date" : "2025-03-18"
 }