diff --git a/connectors-common/file-connector-core/src/main/java/io/tapdata/common/FileConnector.java b/connectors-common/file-connector-core/src/main/java/io/tapdata/common/FileConnector.java index fba3a7b40..5d986ab35 100644 --- a/connectors-common/file-connector-core/src/main/java/io/tapdata/common/FileConnector.java +++ b/connectors-common/file-connector-core/src/main/java/io/tapdata/common/FileConnector.java @@ -255,7 +255,8 @@ protected void makeTapTable(TapTable tapTable, Map sample, boole for (Map.Entry objectEntry : sample.entrySet()) { TapField field = new TapField(); field.name(objectEntry.getKey()); - if (EmptyKit.isNotEmpty((String) objectEntry.getValue()) && ((String) objectEntry.getValue()).length() > 200) { + String value = objectEntry.getValue() == null ? "" : String.valueOf(objectEntry.getValue()); + if (EmptyKit.isNotEmpty(value) && value.length() > 200) { field.dataType("TEXT"); } else { field.dataType("STRING"); @@ -266,7 +267,18 @@ protected void makeTapTable(TapTable tapTable, Map sample, boole for (Map.Entry objectEntry : sample.entrySet()) { TapField field = new TapField(); field.name(objectEntry.getKey()); - String value = (String) objectEntry.getValue(); + Object rawValue = objectEntry.getValue(); + if (rawValue instanceof Map) { + field.dataType("OBJECT"); + tapTable.add(field); + continue; + } + if (rawValue instanceof Collection || (rawValue != null && rawValue.getClass().isArray())) { + field.dataType("ARRAY"); + tapTable.add(field); + continue; + } + String value = rawValue == null ? "" : String.valueOf(rawValue); if (EmptyKit.isEmpty(value)) { field.dataType("STRING"); } else if (MatchUtil.matchBoolean(value)) { diff --git a/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/XmlConnector.java b/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/XmlConnector.java index e58726140..cef7ea1e6 100644 --- a/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/XmlConnector.java +++ b/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/XmlConnector.java @@ -83,19 +83,24 @@ public void registerCapabilities(ConnectorFunctions connectorFunctions, TapCodec @Override public void discoverSchema(TapConnectionContext connectionContext, List tables, int tableSize, Consumer> consumer) throws Throwable { - initConnection(connectionContext); - if (EmptyKit.isBlank(fileConfig.getModelName())) { - return; - } - TapTable tapTable = table(fileConfig.getModelName()); - ConcurrentMap xmlFileMap = getFilteredFiles(); - XmlSchema xmlSchema = new XmlSchema((XmlConfig) fileConfig, storage); - Map sample = xmlSchema.sampleEveryFileData(xmlFileMap); - if (EmptyKit.isEmpty(sample)) { - throw new RuntimeException("Load schema from xml files error: no headers and contents!"); + try { + initConnection(connectionContext); + if (EmptyKit.isBlank(fileConfig.getModelName())) { + return; + } + TapTable tapTable = table(fileConfig.getModelName()); + ConcurrentMap xmlFileMap = getFilteredFiles(); + XmlSchema xmlSchema = new XmlSchema((XmlConfig) fileConfig, storage); + Map sample = xmlSchema.sampleEveryFileData(xmlFileMap); + if (EmptyKit.isEmpty(sample)) { + throw new RuntimeException("Load schema from xml files error: no headers and contents!"); + } + makeTapTable(tapTable, sample, fileConfig.getJustString()); + consumer.accept(Collections.singletonList(tapTable)); + } finally { + if (null != storage) { + storage.destroy(); + } } - makeTapTable(tapTable, sample, fileConfig.getJustString()); - consumer.accept(Collections.singletonList(tapTable)); - storage.destroy(); } } diff --git a/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/handler/BigSaxDataHandler.java b/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/handler/BigSaxDataHandler.java index 31664e5f8..81a823824 100644 --- a/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/handler/BigSaxDataHandler.java +++ b/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/handler/BigSaxDataHandler.java @@ -14,8 +14,6 @@ import org.dom4j.tree.DefaultElement; import org.dom4j.tree.DefaultText; -import java.util.ArrayList; -import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicReference; @@ -26,7 +24,7 @@ import static io.tapdata.entity.simplify.TapSimplify.insertRecordEvent; import static io.tapdata.entity.simplify.TapSimplify.list; -public class BigSaxDataHandler implements ElementHandler { +public class BigSaxDataHandler implements ElementHandler, HandlerBase { private String path; private FileOffset fileOffset; @@ -109,7 +107,8 @@ public void onEnd(ElementPath elementPath) { } } - private Object analyzeElement(Element element) { + @Override + public Object analyzeElement(Element element) { List nodes = element.content(); if (nodes.size() == 1 && nodes.get(0) instanceof DefaultText) { try { @@ -117,17 +116,31 @@ private Object analyzeElement(Element element) { } catch (Exception e) { throw new RuntimeException(String.format("%s field has invalid value", element.getName()), e); } - } else { - List newNodes = nodes.stream().filter(v -> v instanceof DefaultElement).collect(Collectors.toList()); - if (newNodes.stream().map(Node::getPath).distinct().count() > 1) { - Map subMap = new LinkedHashMap<>(); - newNodes.forEach(v -> subMap.put(v.getName(), analyzeElement((DefaultElement) v))); - return subMap; - } else { - List subList = new ArrayList<>(); - newNodes.forEach(v -> subList.add(analyzeElement((DefaultElement) v))); - return subList; + } + List newNodes = nodes.stream().filter(v -> v instanceof DefaultElement).collect(Collectors.toList()); + if (newNodes.isEmpty()) { + // No child elements, but the content may be split into multiple nodes (whitespace text, + // CDATA sections, leading/trailing newlines) by the SAX parser. We must only append + // TEXT_NODE and CDATA_SECTION_NODE to keep semantics consistent with the single-DefaultText + // path above. Explicitly skip XML comments (DefaultComment), processing instructions + // (DefaultProcessingInstruction) and entity references -- those are XML-level metadata + // and must never pollute the downstream business value, which would otherwise turn + // "helloworld" into "hello NOTE world" and silently break type inference, + // regex validations and hash-based idempotency checks on the sink side. + StringBuilder sb = new StringBuilder(); + for (Node node : nodes) { + short nodeType = node.getNodeType(); + if (Node.TEXT_NODE == nodeType || Node.CDATA_SECTION_NODE == nodeType) { + sb.append(node.getText()); + } + } + String text = sb.toString(); + try { + return MatchUtil.parse(text, dataTypeMap.get(element.getName())); + } catch (Exception e) { + throw new RuntimeException(String.format("%s field has invalid value", element.getName()), e); } } + return afterAnalyzeElement(newNodes); } } diff --git a/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/handler/BigSaxSchemaHandler.java b/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/handler/BigSaxSchemaHandler.java index 4e9b5939c..04b85e4c2 100644 --- a/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/handler/BigSaxSchemaHandler.java +++ b/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/handler/BigSaxSchemaHandler.java @@ -8,13 +8,11 @@ import org.dom4j.tree.DefaultElement; import org.dom4j.tree.DefaultText; -import java.util.ArrayList; -import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.stream.Collectors; -public class BigSaxSchemaHandler implements ElementHandler { +public class BigSaxSchemaHandler implements ElementHandler, HandlerBase { private String path; private Map sampleResult; @@ -67,21 +65,30 @@ public void onEnd(ElementPath elementPath) { } } - private Object analyzeElement(Element element) { + @Override + public Object analyzeElement(Element element) { List nodes = element.content(); if (nodes.size() == 1 && nodes.get(0) instanceof DefaultText) { return nodes.get(0).getText(); - } else { - List newNodes = nodes.stream().filter(v -> v instanceof DefaultElement).collect(Collectors.toList()); - if (newNodes.stream().map(Node::getPath).distinct().count() > 1) { - Map subMap = new LinkedHashMap<>(); - newNodes.forEach(v -> subMap.put(v.getName(), analyzeElement((DefaultElement) v))); - return subMap; - } else { - List subList = new ArrayList<>(); - newNodes.forEach(v -> subList.add(analyzeElement((DefaultElement) v))); - return subList; + } + List newNodes = nodes.stream().filter(v -> v instanceof DefaultElement).collect(Collectors.toList()); + if (newNodes.isEmpty()) { + // No child elements, but possibly multiple content nodes due to whitespace/newlines/CDATA splitting + // by the dom4j parser. Only concatenate TEXT and CDATA_SECTION nodes so that the result stays + // semantically aligned with the single-DefaultText branch above. XML comments (DefaultComment), + // processing instructions (DefaultProcessingInstruction) and entity references are metadata, + // so their text must NOT leak into the extracted business value -- otherwise mixed content + // like "helloworld" would incorrectly yield "hello NOTE world" instead of + // "helloworld", and PIs such as "foobar" would incorrectly append "baz" to data. + StringBuilder sb = new StringBuilder(); + for (Node node : nodes) { + short nodeType = node.getNodeType(); + if (Node.TEXT_NODE == nodeType || Node.CDATA_SECTION_NODE == nodeType) { + sb.append(node.getText()); + } } + return sb.toString(); } + return afterAnalyzeElement(newNodes); } } diff --git a/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/handler/HandlerBase.java b/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/handler/HandlerBase.java new file mode 100644 index 000000000..467704706 --- /dev/null +++ b/connectors/xml-connector/src/main/java/io/tapdata/connector/xml/handler/HandlerBase.java @@ -0,0 +1,27 @@ +package io.tapdata.connector.xml.handler; + +import org.dom4j.Element; +import org.dom4j.Node; +import org.dom4j.tree.DefaultElement; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +public interface HandlerBase { + + default Object afterAnalyzeElement(List newNodes) { + if (newNodes.stream().map(Node::getPath).distinct().count() > 1) { + Map subMap = new LinkedHashMap<>(); + newNodes.forEach(v -> subMap.put(v.getName(), analyzeElement((DefaultElement) v))); + return subMap; + } else { + List subList = new ArrayList<>(); + newNodes.forEach(v -> subList.add(analyzeElement((DefaultElement) v))); + return subList; + } + } + + Object analyzeElement(Element element); +}