Skip to content

Fix Snowflake external table 0-row bug with flat JSON array format #328

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
concaf wants to merge 4 commits into
base: main
Choose a base branch
from
Draft

Fix Snowflake external table 0-row bug with flat JSON array format #328

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b30fe29
Fix Snowflake external table 0-row bug by switching to flat JSON arra…
concaf Aug 19, 2026
20bb797
Add nolint:nilerr directives for intentional error swallowing
concaf Aug 21, 2026
c20d323
Move nolint:nilerr directives to the return statement
concaf Aug 21, 2026
7a22f1e
Remove unused nolint directive from embeddings controller
concaf Aug 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 61 additions & 41 deletions internal/controller/chunksgenerator_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,14 +189,14 @@ func (r *ChunksGeneratorReconciler) processConvertedFile(ctx context.Context, co
}

// chunk the file
chunksFile, err := r.chunkFile(ctx, convertedFilePath, chunksGeneratorCR)
chunkRows, err := r.chunkFile(ctx, convertedFilePath, chunksGeneratorCR)
if err != nil {
logger.Error(err, "failed to chunk file")
return false, err
}

// store the chunks in the filestore
chunksFileBytes, err := json.Marshal(chunksFile)
chunksFileBytes, err := json.Marshal(chunkRows)
if err != nil {
logger.Error(err, "failed to marshal chunks file")
return false, err
Expand All @@ -215,21 +215,12 @@ func (r *ChunksGeneratorReconciler) needsChunking(ctx context.Context, converted

chunksFilePath := unstructured.RemapToOutputDir(convertedFilePath, inputPath, outputPath)

// fetch the converted file from the filestore
// this will also make sure that the converted file exists in the filestore
convertedFileRaw, err := r.fileStore.Retrieve(ctx, convertedFilePath)
if err != nil {
return false, err
}

convertedFile := unstructured.ConvertedFile{}
err = json.Unmarshal(convertedFileRaw, &convertedFile)
// fetch the converted file metadata
_, convertedFileMetadata, err := r.readConvertedFile(ctx, convertedFilePath)
if err != nil {
return false, err
}

convertedFileMetadata := convertedFile.ConvertedDocument.Metadata

// check if the chunked file does not exist in the filestore then return true
chunksFileExists, err := r.fileStore.Exists(ctx, chunksFilePath)
if err != nil {
Expand All @@ -245,18 +236,31 @@ func (r *ChunksGeneratorReconciler) needsChunking(ctx context.Context, converted
return false, err
}

chunksFile := unstructured.ChunksFile{}
err = json.Unmarshal(chunksFileRaw, &chunksFile)
if err != nil {
return false, err
}

// now the chunks file should be the same as the current chunks file in filestore
newChunksFileMetadata := unstructured.ChunksFileMetadata{
ConvertedFileMetadata: convertedFileMetadata,
ChunkingTool: unstructured.LangchainChunkingTool,
ChunksGeneratorConfig: chunksGeneratorCR.Spec.ChunksGeneratorConfig,
}

// try new array format first
var chunkRows []unstructured.ChunkRow
if err := json.Unmarshal(chunksFileRaw, &chunkRows); err == nil && len(chunkRows) > 0 && chunkRows[0].Metadata != nil {
if chunkRows[0].Metadata.Equal(&newChunksFileMetadata) {
return false, nil
}
logger.Info("chunks file config has changed, re-chunking needed", "file", convertedFilePath)
return true, nil
}

// fall back to old single-object format
chunksFile := unstructured.ChunksFile{}
if parseErr := json.Unmarshal(chunksFileRaw, &chunksFile); parseErr != nil {
logger.Info("chunks file exists but cannot be parsed, re-chunking needed", "file", convertedFilePath)
return true, nil //nolint:nilerr // unparseable file means re-chunking is needed
}
if chunksFile.ChunksDocument == nil || chunksFile.ChunksDocument.Metadata == nil {
return true, nil
}
if !chunksFile.ChunksDocument.Metadata.Equal(&newChunksFileMetadata) {
logger.Info("chunks file config has changed, re-chunking needed", "file", convertedFilePath)
return true, nil
Expand All @@ -265,18 +269,31 @@ func (r *ChunksGeneratorReconciler) needsChunking(ctx context.Context, converted
return false, nil
}

func (r *ChunksGeneratorReconciler) chunkFile(ctx context.Context, convertedFilePath string, chunksGeneratorCR *operatorv1alpha1.ChunksGenerator) (*unstructured.ChunksFile, error) {
logger := log.FromContext(ctx)
logger.Info("chunking file", "file", convertedFilePath)

// read the converted file from the filestore
func (r *ChunksGeneratorReconciler) readConvertedFile(ctx context.Context, convertedFilePath string) (string, *unstructured.ConvertedFileMetadata, error) {
convertedFileRaw, err := r.fileStore.Retrieve(ctx, convertedFilePath)
if err != nil {
return nil, err
return "", nil, err
}

// try new array format first
var convertedRows []unstructured.ConvertedRow
if err := json.Unmarshal(convertedFileRaw, &convertedRows); err == nil && len(convertedRows) > 0 && convertedRows[0].Metadata != nil {
return convertedRows[0].Markdown, convertedRows[0].Metadata, nil
}

// fall back to old single-object format
convertedFile := unstructured.ConvertedFile{}
err = json.Unmarshal(convertedFileRaw, &convertedFile)
if err := json.Unmarshal(convertedFileRaw, &convertedFile); err != nil {
return "", nil, err
}
return convertedFile.ConvertedDocument.Content.Markdown, convertedFile.ConvertedDocument.Metadata, nil
}

func (r *ChunksGeneratorReconciler) chunkFile(ctx context.Context, convertedFilePath string, chunksGeneratorCR *operatorv1alpha1.ChunksGenerator) ([]unstructured.ChunkRow, error) {
logger := log.FromContext(ctx)
logger.Info("chunking file", "file", convertedFilePath)

markdown, convertedMetadata, err := r.readConvertedFile(ctx, convertedFilePath)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -321,24 +338,27 @@ func (r *ChunksGeneratorReconciler) chunkFile(ctx context.Context, convertedFile
return nil, fmt.Errorf("invalid strategy: %s", chunksGeneratorCR.Spec.ChunksGeneratorConfig.Strategy)
}

chunks, err := chunker.Chunk(convertedFile.ConvertedDocument.Content.Markdown)
chunks, err := chunker.Chunk(markdown)
if err != nil {
return nil, err
}

return &unstructured.ChunksFile{
ConvertedDocument: convertedFile.ConvertedDocument,
ChunksDocument: &unstructured.ChunksDocument{
Metadata: &unstructured.ChunksFileMetadata{
ChunkingTool: unstructured.LangchainChunkingTool,
ChunksGeneratorConfig: chunksGeneratorCR.Spec.ChunksGeneratorConfig,
ConvertedFileMetadata: convertedFile.ConvertedDocument.Metadata,
},
Chunks: &unstructured.Chunks{
Text: chunks,
},
},
}, nil
fileID := convertedMetadata.FileIdentifier
metadata := &unstructured.ChunksFileMetadata{
ChunkingTool: unstructured.LangchainChunkingTool,
ChunksGeneratorConfig: chunksGeneratorCR.Spec.ChunksGeneratorConfig,
ConvertedFileMetadata: convertedMetadata,
}
rows := make([]unstructured.ChunkRow, len(chunks))
for i, text := range chunks {
rows[i] = unstructured.ChunkRow{
FileID: fileID,
ChunkIndex: i,
Text: text,
Metadata: metadata,
}
}
return rows, nil
}

func (r *ChunksGeneratorReconciler) findDependents(ctx context.Context, obj client.Object) []reconcile.Request {
Expand Down
41 changes: 22 additions & 19 deletions internal/controller/documentprocessor_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,16 +232,13 @@ func (r *DocumentProcessorReconciler) reconcileJob(ctx context.Context, job oper
DocumentConverter: unstructured.DocumentConverterDocling,
DoclingConfig: documentProcessorCR.Spec.DocumentProcessorConfig.DoclingConfig,
}
convertedFile := unstructured.ConvertedFile{
ConvertedDocument: &unstructured.ConvertedDocument{
Metadata: &convertedFileMetadata,
Content: &unstructured.Content{
Markdown: doclingResponse.Document.MDContent,
},
},
}
convertedRows := []unstructured.ConvertedRow{{
FileID: job.FileIdentifier,
Markdown: doclingResponse.Document.MDContent,
Metadata: &convertedFileMetadata,
}}

convertedFileBytes, err := json.Marshal(convertedFile)
convertedFileBytes, err := json.Marshal(convertedRows)
if err != nil {
return err
}
Expand Down Expand Up @@ -400,23 +397,29 @@ func (r *DocumentProcessorReconciler) needsConversion(ctx context.Context, rawFi
return false, err
}

convertedFile := unstructured.ConvertedFile{}
err = json.Unmarshal(convertedFileRaw, &convertedFile)
if err != nil {
return false, err
}
currentConvertedFileMetadata := convertedFile.ConvertedDocument.Metadata

fileToConvertMetadata := unstructured.ConvertedFileMetadata{
RawFilePath: rawFilePath,
FileIdentifier: fileUID,
DocumentConverter: unstructured.DocumentConverterDocling,
DoclingConfig: documentProcessorCR.Spec.DocumentProcessorConfig.DoclingConfig,
}

if currentConvertedFileMetadata.Equal(&fileToConvertMetadata) {
logger.Info("converted file has the same configuration, no conversion needed", "filePath", rawFilePath)
return false, nil
// try new array format first
var convertedRows []unstructured.ConvertedRow
if err := json.Unmarshal(convertedFileRaw, &convertedRows); err == nil && len(convertedRows) > 0 && convertedRows[0].Metadata != nil {
if convertedRows[0].Metadata.Equal(&fileToConvertMetadata) {
logger.Info("converted file has the same configuration, no conversion needed", "filePath", rawFilePath)
return false, nil
}
} else {
// fall back to old single-object format
convertedFile := unstructured.ConvertedFile{}
if err := json.Unmarshal(convertedFileRaw, &convertedFile); err == nil && convertedFile.ConvertedDocument != nil && convertedFile.ConvertedDocument.Metadata != nil {
if convertedFile.ConvertedDocument.Metadata.Equal(&fileToConvertMetadata) {
logger.Info("converted file has the same configuration, no conversion needed", "filePath", rawFilePath)
return false, nil
}
}
}
}

Expand Down
Loading
Loading