cherry-pick(1.12.7): fix(csv): correct entityType in recursive import extension validation + row-count accounting (#27593)

openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/DatabaseServiceResourceIT.java

@@ -979,4 +979,259 @@ private String addDomainsToTableRow(String csvLine, String newDomains) {
     }
     return String.join(",", parts);
   }
+
+  @Test
+  void test_recursiveImportCustomPropertyExtension(TestNamespace ns)
+      throws IOException, InterruptedException {
+    String propName = ns.prefix("potato");
+    String serverUrl = SdkClients.getServerUrl();
+    String token = SdkClients.getAdminToken();
+    com.fasterxml.jackson.databind.ObjectMapper mapper =
+        new com.fasterxml.jackson.databind.ObjectMapper();
+    HttpClient client = HttpClient.newHttpClient();
+
+    HttpRequest getStringTypeReq =
+        HttpRequest.newBuilder()
+            .uri(URI.create(serverUrl + "/v1/metadata/types/name/string"))
+            .header("Authorization", "Bearer " + token)
+            .GET()
+            .build();
+    HttpResponse<String> stringTypeResp =
+        client.send(getStringTypeReq, HttpResponse.BodyHandlers.ofString());
+    assertEquals(200, stringTypeResp.statusCode(), "Should fetch string type");
+
+    HttpRequest getTableTypeReq =
+        HttpRequest.newBuilder()
+            .uri(URI.create(serverUrl + "/v1/metadata/types/name/table"))
+            .header("Authorization", "Bearer " + token)
+            .GET()
+            .build();
+    HttpResponse<String> tableTypeResp =
+        client.send(getTableTypeReq, HttpResponse.BodyHandlers.ofString());
+    assertEquals(200, tableTypeResp.statusCode(), "Should fetch table type");
+
+    com.fasterxml.jackson.databind.JsonNode stringTypeNode = mapper.readTree(stringTypeResp.body());
+    com.fasterxml.jackson.databind.JsonNode tableTypeNode = mapper.readTree(tableTypeResp.body());
+    String tableTypeId = tableTypeNode.get("id").asText();
+
+    java.util.Map<String, Object> propertyTypeRef =
+        java.util.Map.of(
+            "id", stringTypeNode.get("id").asText(),
+            "type", "type",
+            "name", stringTypeNode.get("name").asText(),
+            "fullyQualifiedName", stringTypeNode.get("fullyQualifiedName").asText());
+    String customPropertyBody =
+        mapper.writeValueAsString(
+            java.util.Map.of(
+                "name",
+                propName,
+                "description",
+                "Test extension property for recursive import",
+                "propertyType",
+                propertyTypeRef));
+
+    HttpRequest registerPropReq =
+        HttpRequest.newBuilder()
+            .uri(URI.create(serverUrl + "/v1/metadata/types/" + tableTypeId))
+            .header("Authorization", "Bearer " + token)
+            .header("Content-Type", "application/json")
+            .PUT(HttpRequest.BodyPublishers.ofString(customPropertyBody))
+            .build();
+    HttpResponse<String> registerResp =
+        client.send(registerPropReq, HttpResponse.BodyHandlers.ofString());
+    assertEquals(200, registerResp.statusCode(), "Should register custom property on table type");
+
+    try {
+      DatabaseService service =
+          createEntity(createMinimalRequest(ns).withName(ns.prefix("ext_svc")));
+      Database database =
+          SdkClients.adminClient()
+              .databases()
+              .create(
+                  new CreateDatabase()
+                      .withName(ns.prefix("ext_db"))
+                      .withService(service.getFullyQualifiedName()));
+      DatabaseSchema schema =
+          SdkClients.adminClient()
+              .databaseSchemas()
+              .create(
+                  new CreateDatabaseSchema()
+                      .withName(ns.prefix("ext_schema"))
+                      .withDatabase(database.getFullyQualifiedName()));
+
+      String tableName = ns.prefix("ext_tbl");
+      String tableFqn = schema.getFullyQualifiedName() + "." + tableName;
+
+      String validCsv =
+          buildRecursiveCsv(
+              database, schema, tableName, tableFqn, "", propName + ":s3://bucket/file.csv");
+      CsvImportResult validResult =
+          importCsvRecursive(service.getFullyQualifiedName(), validCsv, true);
+      assertEquals(ApiStatus.SUCCESS, validResult.getStatus(), validResult.getImportResultsCsv());
+      assertEquals(0, validResult.getNumberOfRowsFailed());
+      assertEquals(3, validResult.getNumberOfRowsProcessed());
+      assertEquals(3, validResult.getNumberOfRowsPassed());
+
+      String badExtCsv =
+          buildRecursiveCsv(
+              database, schema, tableName, tableFqn, "", "unknown_prop_xyz_test:somevalue");
+      CsvImportResult badResult =
+          importCsvRecursive(service.getFullyQualifiedName(), badExtCsv, true);
+      assertEquals(ApiStatus.PARTIAL_SUCCESS, badResult.getStatus());
+      assertEquals(1, badResult.getNumberOfRowsFailed());
+      assertEquals(3, badResult.getNumberOfRowsProcessed());
+      assertEquals(2, badResult.getNumberOfRowsPassed());
+
+      String dedupCsv =
+          buildRecursiveCsv(
+              database,
+              schema,
+              tableName,
+              tableFqn,
+              "invalidownerformat",
+              "unknown_prop_xyz_test:somevalue");
+      CsvImportResult dedupResult =
+          importCsvRecursive(service.getFullyQualifiedName(), dedupCsv, true);
+      assertEquals(
+          1,
+          dedupResult.getNumberOfRowsFailed(),
+          "Multi-field failure on one row must count as 1 failed row");
+
+    } finally {
+      removeCustomPropertyFromType(tableTypeId, propName, token);
+    }
+  }
+
+  private String buildRecursiveCsv(
+      Database database,
+      DatabaseSchema schema,
+      String tableName,
+      String tableFqn,
+      String tableOwner,
+      String tableExtension) {
+    String header =
+        "name*,displayName,description,owner,tags,glossaryTerms,tiers,certification,"
+            + "retentionPeriod,sourceUrl,domains,extension,entityType*,fullyQualifiedName,"
+            + "column.dataTypeDisplay,column.dataType,column.arrayDataType,column.dataLength,"
+            + "storedProcedure.code,storedProcedure.language";
+    String dbRow =
+        csvRow(
+            database.getName(),
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "database",
+            database.getFullyQualifiedName(),
+            "",
+            "",
+            "",
+            "",
+            "",
+            "");
+    String schemaRow =
+        csvRow(
+            schema.getName(),
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "databaseSchema",
+            schema.getFullyQualifiedName(),
+            "",
+            "",
+            "",
+            "",
+            "",
+            "");
+    String tableRow =
+        csvRow(
+            tableName,
+            "",
+            "",
+            tableOwner,
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "",
+            tableExtension,
+            "table",
+            tableFqn,
+            "",
+            "",
+            "",
+            "",
+            "",
+            "");
+    return header + "\n" + dbRow + "\n" + schemaRow + "\n" + tableRow + "\n";
+  }
+
+  private void removeCustomPropertyFromType(String typeId, String propName, String token)
+      throws IOException, InterruptedException {
+    com.fasterxml.jackson.databind.ObjectMapper localMapper =
+        new com.fasterxml.jackson.databind.ObjectMapper();
+    HttpClient client = HttpClient.newHttpClient();
+    String baseUrl = SdkClients.getServerUrl();
+    String getUrl = baseUrl + "/v1/metadata/types/" + typeId + "?fields=customProperties";
+    HttpRequest getReq =
+        HttpRequest.newBuilder()
+            .uri(URI.create(getUrl))
+            .header("Authorization", "Bearer " + token)
+            .GET()
+            .build();
+    HttpResponse<String> getResp = client.send(getReq, HttpResponse.BodyHandlers.ofString());
+    if (getResp.statusCode() != 200) {
+      return;
+    }
+    com.fasterxml.jackson.databind.JsonNode typeNode = localMapper.readTree(getResp.body());
+    com.fasterxml.jackson.databind.JsonNode customProps = typeNode.get("customProperties");
+    if (customProps == null || !customProps.isArray()) {
+      return;
+    }
+    for (int i = 0; i < customProps.size(); i++) {
+      if (propName.equals(customProps.get(i).path("name").asText())) {
+        String patchBody = "[{\"op\":\"remove\",\"path\":\"/customProperties/" + i + "\"}]";
+        HttpRequest patchReq =
+            HttpRequest.newBuilder()
+                .uri(URI.create(baseUrl + "/v1/metadata/types/" + typeId))
+                .header("Authorization", "Bearer " + token)
+                .header("Content-Type", "application/json-patch+json")
+                .method("PATCH", HttpRequest.BodyPublishers.ofString(patchBody))
+                .build();
+        client.send(patchReq, HttpResponse.BodyHandlers.ofString());
+        break;
+      }
+    }
+  }
+
+  private String csvRow(String... fields) {
+    StringBuilder sb = new StringBuilder();
+    for (int i = 0; i < fields.length; i++) {
+      if (i > 0) sb.append(",");
+      String field = fields[i];
+      if (field.contains(",") || field.contains("\"") || field.contains("\n")) {
+        sb.append('"').append(field.replace("\"", "\"\"")).append('"');
+      } else {
+        sb.append(field);
+      }
+    }
+    return sb.toString();
+  }
 }

openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/GlossaryResourceIT.java

@@ -912,11 +912,8 @@ void test_bulkImportGlossaryTermsIncrementsVersion(TestNamespace ns) {
       importResult = JsonUtils.readValue(result, CsvImportResult.class);
       assertNotNull(importResult, "Should parse CsvImportResult from response");
       assertEquals(ApiStatus.SUCCESS, importResult.getStatus(), "Import should succeed");
-      // numberOfRowsProcessed = header row (1) + 3 data rows = 4
-      assertEquals(
-          4, importResult.getNumberOfRowsProcessed(), "Should process 4 rows (header + 3 data)");
-      assertEquals(
-          4, importResult.getNumberOfRowsPassed(), "All 4 rows should pass (header + 3 data)");
+      assertEquals(3, importResult.getNumberOfRowsProcessed(), "Should process 3 data rows");
+      assertEquals(3, importResult.getNumberOfRowsPassed(), "All 3 data rows should pass");
       assertEquals(0, importResult.getNumberOfRowsFailed(), "No rows should fail");
       assertFalse(importResult.getDryRun(), "Should not be a dry run");
     } catch (Exception e) {

openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/TestCaseResourceIT.java

@@ -2838,13 +2838,13 @@ void test_importCsvWithWildcardName_multipleTablesSucceeds(TestNamespace ns) {
     // Dry run with name="*" should succeed
     CsvImportResult dryRunResult = importCsvWithWildcard(client, csvData, true);
     assertEquals(ApiStatus.SUCCESS, dryRunResult.getStatus());
-    assertEquals(3, dryRunResult.getNumberOfRowsProcessed());
+    assertEquals(2, dryRunResult.getNumberOfRowsProcessed());
 
     // Actual import with name="*" — previously failed because
     // processChangeEventForBulkImport would call getByName("*")
     CsvImportResult result = importCsvWithWildcard(client, csvData, false);
     assertEquals(ApiStatus.SUCCESS, result.getStatus());
-    assertEquals(3, result.getNumberOfRowsProcessed());
+    assertEquals(2, result.getNumberOfRowsProcessed());
 
     // Verify test cases created on different tables
     TestCase tc1 =
@@ -2905,7 +2905,7 @@ void test_importCsvWithWildcardName_explicitTestSuiteTracked(TestNamespace ns) {
 
     CsvImportResult result = importCsvWithWildcard(client, csvData, false);
     assertEquals(ApiStatus.SUCCESS, result.getStatus());
-    assertEquals(2, result.getNumberOfRowsProcessed());
+    assertEquals(1, result.getNumberOfRowsProcessed());
 
     TestCase imported =
         client.testCases().getByName(table.getFullyQualifiedName() + "." + testName, "testSuite");
@@ -2966,7 +2966,7 @@ void test_importCsvWithWildcardName_dryRunDoesNotCreateEntities(TestNamespace ns
 
     CsvImportResult dryRunResult = importCsvWithWildcard(client, csvData, true);
     assertEquals(ApiStatus.SUCCESS, dryRunResult.getStatus());
-    assertEquals(2, dryRunResult.getNumberOfRowsProcessed());
+    assertEquals(1, dryRunResult.getNumberOfRowsProcessed());
 
     // Entity should NOT exist after dry run
     String expectedFqn = table.getFullyQualifiedName() + "." + testName;

openmetadata-service/src/main/java/org/openmetadata/csv/CsvUtil.java

@@ -353,12 +353,12 @@ public static List<String> addExtension(List<String> csvRecord, Object extension
 
     String extensionString =
         extensionMap.entrySet().stream()
+            .map(entry -> Map.entry(entry.getKey(), formatValue(entry.getValue())))
+            .filter(entry -> !entry.getValue().isBlank())
             .map(
-                entry -> {
-                  String key = entry.getKey();
-                  Object value = entry.getValue();
-                  return CsvUtil.quoteCsvField(key + ENTITY_TYPE_SEPARATOR + formatValue(value));
-                })
+                entry ->
+                    CsvUtil.quoteCsvField(
+                        entry.getKey() + ENTITY_TYPE_SEPARATOR + entry.getValue()))
             .collect(Collectors.joining(FIELD_SEPARATOR));
 
     csvRecord.add(extensionString);