Skip to content

Commit c15357b

Browse files
cmgrotecmgrote
authored
Merge pull request #2601 from atlanhq/owen/csa-484
fix(csv): use configured field separator in parallel chunk path
2 parents 80ea684 + d9726b5 commit c15357b

11 files changed

Lines changed: 454 additions & 24 deletions

File tree

‎package-toolkit/runtime/src/main/kotlin/com/atlan/pkg/serde/csv/CSVReader.kt‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ class CSVReader
5959
private val atlanTagHandling: AtlanTagHandling = AtlanTagHandling.REPLACE,
6060
private val creationHandling: AssetCreationHandling = AssetCreationHandling.FULL,
6161
private val tableViewAgnostic: Boolean = false,
62-
fieldSeparator: Char = ',',
62+
private val fieldSeparator: Char = ',',
6363
private val linkIdempotency: LinkIdempotencyInvariant = LinkIdempotencyInvariant.URL,
6464
) : Closeable {
6565
private val reader: CsvReader<CsvRecord>
@@ -187,7 +187,7 @@ class CSVReader
187187
val chunkSize = totalRowCount / parallelism
188188
val currentRecordCount = AtomicLong(0)
189189
var chunkPath = Files.createTempFile("chunk_0_", ".csv")
190-
var writer = CSVWriter(chunkPath.toString(), ',')
190+
var writer = CSVWriter(chunkPath.toString(), fieldSeparator)
191191
reader.stream().skip(1).forEach { row: CsvRecord ->
192192
// Split the original file up into multiple smaller files for parallel-processing
193193
if (rowToAsset.includeRow(row.fields, header, typeIdx, qualifiedNameIdx)) {
@@ -199,7 +199,7 @@ class CSVReader
199199
csvChunkFiles.add(chunkPath)
200200
currentRecordCount.set(0)
201201
chunkPath = Files.createTempFile("chunk_${csvChunkFiles.size}_", ".csv")
202-
writer = CSVWriter(chunkPath.toString(), ',')
202+
writer = CSVWriter(chunkPath.toString(), fieldSeparator)
203203
}
204204
}
205205
}
@@ -216,7 +216,7 @@ class CSVReader
216216
val reader =
217217
CsvReader
218218
.builder()
219-
.fieldSeparator(',')
219+
.fieldSeparator(fieldSeparator)
220220
.quoteCharacter('"')
221221
.skipEmptyLines(true)
222222
.extraFieldStrategy(FieldMismatchStrategy.STRICT)

‎package-toolkit/runtime/src/main/kotlin/com/atlan/pkg/util/DeltaProcessor.kt‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ import java.nio.file.Paths
3131
* @param previousFilePreprocessor responsible for pre-processing the previous CSV file (if provided directly)
3232
* @param outputDirectory local directory where files can be written and compared
3333
* @param previousFileProcessedExtension extension to use in the object store for files that have been processed
34+
* @param fieldSeparator character used to separate fields in the CSV files being compared (for example ',' or ';')
3435
*/
3536
class DeltaProcessor(
3637
val ctx: PackageContext<*>,
@@ -46,6 +47,7 @@ class DeltaProcessor(
4647
val previousFilePreprocessor: CSVPreprocessor? = null,
4748
val outputDirectory: String = Paths.get(separator, "tmp").toString(),
4849
private val previousFileProcessedExtension: String = ".processed",
50+
private val fieldSeparator: Char = ',',
4951
) : AtlanCloseable {
5052
private val objectStore = Utils.getBackingStore(outputDirectory)
5153
private var initialLoad: Boolean = true
@@ -91,6 +93,7 @@ class DeltaProcessor(
9193
purgeAssets,
9294
!reloadAll,
9395
outputDirectory,
96+
fieldSeparator,
9497
)
9598
delta!!.calculateDelta(preprocessedDetails.preprocessedFile, previousFile)
9699
} else {

‎package-toolkit/runtime/src/main/kotlin/com/atlan/pkg/util/FileBasedDelta.kt‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ import kotlin.streams.asSequence
4545
* @param purge if true, any asset that matches will be permanently deleted (otherwise, default: only archived)
4646
* @param compareChecksums if true, compare the checksums of every asset identity to determine which have changed (otherwise, default: skip checksum comparisons)
4747
* @param fallback directory to use as a fallback backing store (locally) in the absence of an object store
48+
* @param fieldSeparator character used to separate fields in the CSV files being compared (for example ',' or ';')
4849
*/
4950
class FileBasedDelta(
5051
private val ctx: PackageContext<*>,
@@ -55,6 +56,7 @@ class FileBasedDelta(
5556
private val purge: Boolean = false,
5657
private val compareChecksums: Boolean = false,
5758
private val fallback: String = Paths.get(separator, "tmp").toString(),
59+
private val fieldSeparator: Char = ',',
5860
) : AtlanCloseable {
5961
val assetsToReload = ChecksumCache("changes")
6062
val assetsToDelete = ChecksumCache("deletes")
@@ -160,7 +162,7 @@ class FileBasedDelta(
160162
filename: String,
161163
cache: ChecksumCache,
162164
) {
163-
val header = CSVXformer.getHeader(filename, ',')
165+
val header = CSVXformer.getHeader(filename, fieldSeparator)
164166
val typeIdx = header.indexOf(Asset.TYPE_NAME.atlanFieldName)
165167
if (typeIdx < 0) {
166168
throw IOException(
@@ -171,7 +173,7 @@ class FileBasedDelta(
171173
val builder =
172174
CsvReader
173175
.builder()
174-
.fieldSeparator(',')
176+
.fieldSeparator(fieldSeparator)
175177
.quoteCharacter('"')
176178
.skipEmptyLines(true)
177179
.extraFieldStrategy(FieldMismatchStrategy.STRICT)

‎package-toolkit/runtime/src/test/kotlin/serde/CSVFieldSeparatorTest.kt‎

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
/* SPDX-License-Identifier: Apache-2.0
2+
Copyright 2023 Atlan Pte. Ltd. */
3+
package serde
4+
5+
import de.siegmar.fastcsv.reader.CsvParseException
6+
import de.siegmar.fastcsv.reader.CsvReader
7+
import de.siegmar.fastcsv.reader.CsvRecord
8+
import de.siegmar.fastcsv.reader.FieldMismatchStrategy
9+
import de.siegmar.fastcsv.writer.CsvWriter
10+
import de.siegmar.fastcsv.writer.LineDelimiter
11+
import de.siegmar.fastcsv.writer.QuoteStrategies
12+
import java.nio.file.Files
13+
import kotlin.test.Test
14+
import kotlin.test.assertEquals
15+
import kotlin.test.assertFailsWith
16+
17+
/**
18+
* Proves that reading a semicolon-delimited CSV with a hardcoded comma
19+
* field separator (as CSVReader.streamRows does in its parallel chunk path)
20+
* crashes with CsvParseException.
21+
*
22+
* Bug: CSVReader.streamRows() at lines 190, 202, 219 hardcodes ','
23+
* when splitting and re-reading chunk files, because fieldSeparator
24+
* is a bare constructor param (not a val) and is inaccessible in methods.
25+
*/
26+
class CSVFieldSeparatorTest {
27+
companion object {
28+
private val HEADER = listOf("qualifiedName", "name", "typeName", "description")
29+
private val ROW1 = listOf("default/schema/table/col1", "col1", "Column", "First column")
30+
private val ROW2 = listOf("default/schema/table/col2", "col2", "Column", "Second column")
31+
}
32+
33+
private fun writeCSV(separator: Char): java.nio.file.Path {
34+
val path = Files.createTempFile("csv_sep_test_", ".csv")
35+
val writer =
36+
CsvWriter
37+
.builder()
38+
.fieldSeparator(separator)
39+
.quoteCharacter('"')
40+
.quoteStrategy(QuoteStrategies.NON_EMPTY)
41+
.lineDelimiter(LineDelimiter.LF)
42+
.build(path)
43+
writer.writeRecord(HEADER)
44+
writer.writeRecord(ROW1)
45+
writer.writeRecord(ROW2)
46+
writer.close()
47+
return path
48+
}
49+
50+
private fun readCSV(
51+
path: java.nio.file.Path,
52+
separator: Char,
53+
): List<List<String>> {
54+
val reader =
55+
CsvReader
56+
.builder()
57+
.fieldSeparator(separator)
58+
.quoteCharacter('"')
59+
.skipEmptyLines(true)
60+
.extraFieldStrategy(FieldMismatchStrategy.STRICT)
61+
.missingFieldStrategy(FieldMismatchStrategy.STRICT)
62+
.ofCsvRecord(path)
63+
val rows = mutableListOf<List<String>>()
64+
reader.stream().skip(1).forEach { r: CsvRecord ->
65+
rows.add(r.fields.toList())
66+
}
67+
reader.close()
68+
return rows
69+
}
70+
71+
@Test
72+
fun semicolonCSVReadWithCorrectSeparator() {
73+
val path = writeCSV(';')
74+
val rows = readCSV(path, ';')
75+
assertEquals(2, rows.size, "Should read 2 data rows")
76+
assertEquals(4, rows[0].size, "Each row should have 4 fields")
77+
assertEquals("default/schema/table/col1", rows[0][0])
78+
assertEquals("col1", rows[0][1])
79+
assertEquals("Column", rows[0][2])
80+
assertEquals("First column", rows[0][3])
81+
Files.deleteIfExists(path)
82+
}
83+
84+
@Test
85+
fun semicolonCSVReadWithHardcodedCommaCrashes() {
86+
// Simulates what CSVReader.streamRows() does at line 219:
87+
// reads semicolon-delimited chunk files with hardcoded fieldSeparator(',')
88+
val path = writeCSV(';')
89+
assertFailsWith<CsvParseException>(
90+
message = "Reading semicolon CSV with comma separator should throw CsvParseException",
91+
) {
92+
readCSV(path, ',')
93+
}
94+
Files.deleteIfExists(path)
95+
}
96+
97+
@Test
98+
fun commaCSVReadWithCommaSeparatorWorks() {
99+
val path = writeCSV(',')
100+
val rows = readCSV(path, ',')
101+
assertEquals(2, rows.size)
102+
assertEquals(4, rows[0].size)
103+
assertEquals("default/schema/table/col1", rows[0][0])
104+
Files.deleteIfExists(path)
105+
}
106+
}

‎samples/packages/asset-import/src/main/kotlin/com/atlan/pkg/aim/Importer.kt‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -52,8 +52,8 @@ object Importer {
5252

5353
val assetsInput =
5454
if (assetsFileProvided) {
55-
if (assetsFieldSeparator.length > 1) {
56-
logger.error { "Field separator must be only a single character. The provided value is too long: $assetsFieldSeparator" }
55+
if (assetsFieldSeparator.length != 1) {
56+
logger.error { "Field separator must be exactly one character. The provided value is invalid: '$assetsFieldSeparator'" }
5757
exitProcess(2)
5858
}
5959
Utils.getInputFile(
@@ -103,8 +103,8 @@ object Importer {
103103
ctx.config.tagsKey,
104104
)
105105
FieldSerde.FAIL_ON_ERRORS.set(tagsFailOnErrors)
106-
if (tagsFieldSeparator.length > 1) {
107-
logger.error { "Field separator must be only a single character. The provided value is too long: $tagsFieldSeparator" }
106+
if (tagsFieldSeparator.length != 1) {
107+
logger.error { "Field separator must be exactly one character. The provided value is invalid: '$tagsFieldSeparator'" }
108108
exitProcess(2)
109109
}
110110
logger.info { "=== Importing tag definitions... ===" }
@@ -127,8 +127,8 @@ object Importer {
127127
ctx.config.glossariesKey,
128128
)
129129
FieldSerde.FAIL_ON_ERRORS.set(glossariesFailOnErrors)
130-
if (glossariesFieldSeparator.length > 1) {
131-
logger.error { "Field separator must be only a single character. The provided value is too long: $glossariesFieldSeparator" }
130+
if (glossariesFieldSeparator.length != 1) {
131+
logger.error { "Field separator must be exactly one character. The provided value is invalid: '$glossariesFieldSeparator'" }
132132
exitProcess(2)
133133
}
134134
logger.info { "=== Importing glossaries... ===" }
@@ -166,8 +166,8 @@ object Importer {
166166
ctx.config.dataProductsKey,
167167
)
168168
FieldSerde.FAIL_ON_ERRORS.set(dataProductsFailOnErrors)
169-
if (dataProductsFieldSeparator.length > 1) {
170-
logger.error { "Field separator must be only a single character. The provided value is too long: $dataProductsFieldSeparator" }
169+
if (dataProductsFieldSeparator.length != 1) {
170+
logger.error { "Field separator must be exactly one character. The provided value is invalid: '$dataProductsFieldSeparator'" }
171171
exitProcess(2)
172172
}
173173
logger.info { "=== Importing domains... ===" }
@@ -239,6 +239,7 @@ object Importer {
239239
null
240240
},
241241
outputDirectory = outputDirectory,
242+
fieldSeparator = assetsFieldSeparator[0],
242243
).use { delta ->
243244

244245
delta.calculate()

‎samples/packages/asset-import/src/test/kotlin/com/atlan/pkg/aim/EmptyFieldSeparatorTest.kt‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
/* SPDX-License-Identifier: Apache-2.0
2+
Copyright 2023 Atlan Pte. Ltd. */
3+
package com.atlan.pkg.aim
4+
5+
import AssetImportCfg
6+
import com.atlan.pkg.PackageTest
7+
import com.atlan.pkg.Utils
8+
import java.nio.file.Paths
9+
import kotlin.test.Test
10+
import kotlin.test.assertEquals
11+
12+
/**
13+
* Test that an invalid field separator is rejected up-front with a non-zero exit code,
14+
* before any file parsing (which would otherwise blow up on the subsequent `[0]` access).
15+
*
16+
* The guard requires the separator to be exactly one character, so both an empty string
17+
* and a multi-character value must fail with exit code 2. (CSA-484)
18+
*/
19+
class EmptyFieldSeparatorTest : PackageTest("efs") {
20+
override val logger = Utils.getLogger(this.javaClass.name)
21+
22+
private val testFile = "assets.csv"
23+
24+
private fun prepFile() {
25+
// The guard fires before the file is read, so any content is sufficient -- it just
26+
// needs to exist so that a file is considered "provided".
27+
val output = Paths.get(testDirectory, testFile).toFile()
28+
output.writeText("qualifiedName,typeName,name\n")
29+
}
30+
31+
/**
32+
* Run the importer with the provided field separator and return the exit code.
33+
* `assetsConfig = "advanced"` is required for the non-default separator to take effect
34+
* (otherwise `getEffectiveValue` falls back to the default comma).
35+
*/
36+
private fun runWithSeparator(separator: String): Int? {
37+
sysExit.execute {
38+
runCustomPackage(
39+
AssetImportCfg(
40+
assetsFile = Paths.get(testDirectory, testFile).toString(),
41+
assetsUpsertSemantic = "upsert",
42+
assetsConfig = "advanced",
43+
assetsFieldSeparator = separator,
44+
),
45+
Importer::main,
46+
)
47+
}
48+
return sysExit.exitCode
49+
}
50+
51+
override fun setup() {
52+
prepFile()
53+
}
54+
55+
@Test
56+
fun emptyFieldSeparatorExitsWithCode2() {
57+
assertEquals(2, runWithSeparator(""), "An empty field separator should fail with exit code 2")
58+
}
59+
60+
@Test
61+
fun multiCharacterFieldSeparatorExitsWithCode2() {
62+
assertEquals(2, runWithSeparator(";;"), "A multi-character field separator should fail with exit code 2")
63+
}
64+
}

0 commit comments

Comments
 (0)