001package gudusoft.gsqlparser.dlineage.dataflow.metadata;
002
003import gudusoft.gsqlparser.dlineage.dataflow.metadata.model.MetadataRelation;
004import gudusoft.gsqlparser.dlineage.dataflow.model.RelationshipType;
005import gudusoft.gsqlparser.util.SQLUtil;
006import gudusoft.gsqlparser.util.csv.CsvReader;
007
008import java.util.ArrayList;
009import java.util.Arrays;
010import java.util.List;
011
012import static gudusoft.gsqlparser.util.json.JSONUtil.extractTopLevelFieldValue;
013import static gudusoft.gsqlparser.util.json.JSONUtil.extractTopLevelStringField;
014
015public class MetadataReader {
016
017    /**
018     * Highest {@code sqlflow-sharded} {@code formatVersion} this reader understands.
019     * The exporter contract is additive within a major version, so a manifest at or
020     * below this is consumed; a higher one is refused rather than silently mis-parsed.
021     */
022    public static final int SUPPORTED_SHARDED_FORMAT_VERSION = 2;
023
024    /** The {@code "sqlflow-sharded"} manifest format tag. */
025    public static final String SHARDED_FORMAT = "sqlflow-sharded";
026
027    /**
028     * Sentinel for a {@code sqlflow-sharded} manifest whose {@code formatVersion} is
029     * present but unusable (out of int range, non-integral, or non-numeric). It never
030     * passes {@link #isSupportedSqlflowSharded}, so such a manifest is refused rather
031     * than silently accepted via a lossy conversion.
032     */
033    public static final int UNSUPPORTED_SHARDED_VERSION = Integer.MAX_VALUE;
034
035    /**
036     * The declared {@code formatVersion} of a sharded manifest, or {@code 1} when the
037     * field is absent (v1 predates the field). Returns {@code -1} when {@code metadata}
038     * is not a {@code sqlflow-sharded} manifest at all, and
039     * {@link #UNSUPPORTED_SHARDED_VERSION} when it IS a sharded manifest but the version
040     * field cannot be read as a supported integer.
041     */
042    public static int shardedFormatVersion(String metadata) {
043        if (SQLUtil.isEmpty(metadata)) {
044            return -1;
045        }
046        String format;
047        try {
048            format = extractTopLevelStringField(metadata, "format");
049        } catch (Exception e) {
050            return -1;
051        }
052        if (format == null || !SHARDED_FORMAT.equals(format)) {
053            return -1;
054        }
055        // From here it IS a sharded manifest; a bad version is UNSUPPORTED, not -1.
056        String v = extractTopLevelFieldValue(metadata, "formatVersion");
057        if (v == null) {
058            return 1;
059        }
060        // Parse exactly on the textual form, never via double (which rounds
061        // 2.0000000000000001 to 2.0) or a lossy intValue() (which wraps
062        // 4294967297 to 1). A value that is not an integer in [1, MAX-1] is
063        // UNSUPPORTED. MAX_VALUE itself maps to UNSUPPORTED so the sentinel does
064        // not silently accept a real version equal to it.
065        try {
066            java.math.BigInteger bi = new java.math.BigDecimal(v.trim())
067                    .toBigIntegerExact(); // throws if fractional
068            if (bi.compareTo(java.math.BigInteger.ONE) < 0
069                    || bi.compareTo(java.math.BigInteger.valueOf(UNSUPPORTED_SHARDED_VERSION - 1)) > 0) {
070                return UNSUPPORTED_SHARDED_VERSION;
071            }
072            return bi.intValue();
073        } catch (ArithmeticException | NumberFormatException e) {
074            return UNSUPPORTED_SHARDED_VERSION;
075        }
076    }
077
078    /**
079     * True iff {@code metadata} is a {@code sqlflow-sharded} manifest whose
080     * {@code formatVersion} this reader supports (≤ {@link #SUPPORTED_SHARDED_FORMAT_VERSION}).
081     * A manifest with a higher version is a {@code sqlflow-sharded} manifest
082     * ({@link #isSqlflowSharded} is still true) but is NOT supported — the caller must
083     * refuse it, not consume it as if it were the supported version.
084     */
085    public static boolean isSupportedSqlflowSharded(String metadata) {
086        int v = shardedFormatVersion(metadata);
087        return v >= 1 && v <= SUPPORTED_SHARDED_FORMAT_VERSION;
088    }
089
090    public static List<MetadataRelation> read(String metadata) {
091        List<MetadataRelation> relations = new ArrayList<>();
092        CsvReader csvReader = CsvReader.parse(metadata);
093        csvReader.setSkipEmptyRecords(true);
094        csvReader.setTrimWhitespace(true);
095        try {
096            if (csvReader.readHeaders()) {
097                String[] headers = csvReader.getHeaders();
098                boolean fromSqlflow = Arrays.stream(headers).filter(t -> t.toLowerCase().trim().equals("source_table_id")).count() > 0;
099                csvReader.readRecord();
100                while (csvReader.readRecord()) {
101                    if (fromSqlflow) {
102                        MetadataRelation relation = new MetadataRelation();
103                        relation.setSourceDb(csvReader.get(0));
104                        relation.setSourceSchema(csvReader.get(1));
105                        relation.setSourceTable(csvReader.get(3));
106                        relation.setSourceColumn(csvReader.get(5));
107                        relation.setTargetDb(csvReader.get(6));
108                        relation.setTargetSchema(csvReader.get(7));
109                        relation.setTargetTable(csvReader.get(9));
110                        relation.setTargetColumn(csvReader.get(11));
111                        relation.setRelationType(RelationshipType.of(csvReader.get(12)).name());
112                        relations.add(relation);
113                    } else if (headers.length >= 8) {
114                        MetadataRelation relation = new MetadataRelation();
115                        relation.setSourceDb(csvReader.get(0));
116                        relation.setSourceSchema(csvReader.get(1));
117                        relation.setSourceTable(csvReader.get(2));
118                        relation.setSourceColumn(csvReader.get(3));
119                        relation.setTargetDb(csvReader.get(4));
120                        relation.setTargetSchema(csvReader.get(5));
121                        relation.setTargetTable(csvReader.get(6));
122                        relation.setTargetColumn(csvReader.get(7));
123                        if (headers.length >= 9) {
124                            relation.setProcedureName(csvReader.get(8));
125                        }
126                        if (headers.length >= 10) {
127                            relation.setQueryName(csvReader.get(9));
128                        }
129                        relations.add(relation);
130                    }
131                }
132            }
133        } catch (Exception e) {
134
135        } finally {
136            csvReader.close();
137        }
138        return relations;
139    }
140
141    public static boolean isMetadata(String metadata) {
142        if (SQLUtil.isEmpty(metadata)) {
143            return false;
144        }
145        CsvReader csvReader = CsvReader.parse(metadata);
146        csvReader.setSkipEmptyRecords(true);
147        csvReader.setTrimWhitespace(true);
148        try {
149            if (csvReader.readHeaders()) {
150                String[] headers = csvReader.getHeaders();
151                if (headers[0].toLowerCase().trim().startsWith("source_db")) {
152                    return true;
153                }
154            }
155        } catch (Exception e) {
156
157        } finally {
158            csvReader.close();
159        }
160        return false;
161    }
162
163    public static boolean isGrabit(String metadata) {
164        if (SQLUtil.isEmpty(metadata)) {
165            return false;
166        }
167        try {
168            String createdBy = extractTopLevelStringField(metadata, "createdBy");
169            if (createdBy != null && createdBy.toLowerCase().indexOf("grabit") != -1) {
170                return true;
171            }
172        } catch (Exception e) {
173        }
174        return false;
175    }
176
177    /**
178         * Check if the given metadata is in the old sqlflow format.
179         *
180         * <p>The old sqlflow format has no {@code format} key and identifies itself
181         * via {@code createdBy} containing "sqlflow". This is distinct from the newer
182         * {@code sqlflow-sharded} format, which has {@code format: "sqlflow-sharded"}.
183         *
184         * <p>Data structure: {@code servers[].queries[]} — queries are nested under
185         * servers, not at the root level.
186         *
187         * @param metadata the JSON string to check
188         * @return true if the metadata is in the old sqlflow format
189         */
190        public static boolean isSqlflow(String metadata) {
191        if (SQLUtil.isEmpty(metadata)) {
192            return false;
193        }
194        try {
195            if (extractTopLevelStringField(metadata, "format") != null) {
196                return false;
197            }
198            String createdBy = extractTopLevelStringField(metadata, "createdBy");
199            if (createdBy != null && createdBy.toLowerCase().indexOf("sqlflow") != -1) {
200                return true;
201            }
202        } catch (Exception e) {
203        }
204        return false;
205    }
206
207    public static boolean isSqlflowSharded(String metadata) {
208        if (SQLUtil.isEmpty(metadata)) {
209            return false;
210        }
211        try {
212            String format = extractTopLevelStringField(metadata, "format");
213            if ("sqlflow-sharded".equals(format)) {
214                return true;
215            }
216        } catch (Exception e) {
217        }
218        return false;
219    }
220}