CSV Mapping (does not fix ingestion problems)

yogilad · Nov 7, 2023 · 049cb4b · 049cb4b
1 parent 4b1eb4e
commit 049cb4b
Show file tree

Hide file tree

Showing 2 changed files with 47 additions and 5 deletions.
diff --git a/Src/App/Utils/Helpers/TabularDataHelper.cs b/Src/App/Utils/Helpers/TabularDataHelper.cs
@@ -170,15 +170,15 @@ public string ToSchemaString(KqlTypeDetectionMode detectionMode, bool strictEnti
             return schemaBuilder.ToString();
         }
 
-        public IngestionMapping ToJsonMapping()
+        public IngestionMapping ToJsonMapping(KqlTypeDetectionMode detectionMode)
         {
             var mappingList = new List<ColumnMapping>();
 
             for (int i = 0; i < Columns.Count; i++)
             {
                 var col = Columns[i];
-                var colMapping = new ColumnMapping(StrippedColumnName(NormalizeColumnName(col.Name, i)), 
-                    col.Type.Name, 
+                var colMapping = new ColumnMapping(StrippedColumnName(NormalizeColumnName(col.Name, i)),
+                    NormalizeTypeName(col.Type, detectionMode), 
                     new Dictionary<string, string>() { {"Path", $"$.{StrippedColumnName(col.Name)}"} });
 
                 mappingList.Add(colMapping);
@@ -192,6 +192,29 @@ public IngestionMapping ToJsonMapping()
 
             return mapping;
         }
+
+        public IngestionMapping ToCsvMapping(KqlTypeDetectionMode detectionMode)
+        {
+            var mappingList = new List<ColumnMapping>();
+
+            for (int i = 0; i < Columns.Count; i++)
+            {
+                var col = Columns[i];
+                var colMapping = new ColumnMapping(StrippedColumnName(NormalizeColumnName(col.Name, i)),
+                    NormalizeTypeName(col.Type, detectionMode),
+                    new Dictionary<string, string>() { { "Ordinal", i.ToString() } });
+
+                mappingList.Add(colMapping);
+            }
+
+            var mapping = new IngestionMapping()
+            {
+                IngestionMappingKind = Kusto.Data.Ingestion.IngestionMappingKind.Csv,
+                IngestionMappings = mappingList,
+            };
+
+            return mapping;
+        }
     }
     #endregion
 

diff --git a/Src/App/Workers/Ingest/TempTableWorker.cs b/Src/App/Workers/Ingest/TempTableWorker.cs
@@ -194,7 +194,17 @@ private async Task<bool> HandleSingleTextStreamAsync(KustoDatabaseHelper databas
                         case DataSourceFormat.json:
                         case DataSourceFormat.singlejson:
                         case DataSourceFormat.multijson:
-                            mapping = autoDetectRes.Schema.ToJsonMapping();
+                            mapping = autoDetectRes.Schema.ToJsonMapping(detectionMode);
+                            break;
+
+                        case DataSourceFormat.csv:
+                        case DataSourceFormat.psv:
+                        case DataSourceFormat.scsv:
+                        case DataSourceFormat.sohsv:
+                        case DataSourceFormat.tsv:
+                        case DataSourceFormat.tsve:
+                        case DataSourceFormat.w3clogfile:
+                            mapping = autoDetectRes.Schema.ToCsvMapping(detectionMode);
                             break;
                     }
                 }
@@ -217,7 +227,16 @@ private async Task<bool> HandleSingleTextStreamAsync(KustoDatabaseHelper databas
                         case DataSourceFormat.json:
                         case DataSourceFormat.singlejson:
                         case DataSourceFormat.multijson:
-                            mapping = schemaRes.TableScheme.ToJsonMapping();
+                            mapping = schemaRes.TableScheme.ToJsonMapping(detectionMode);
+                            break;
+
+                        case DataSourceFormat.csv:
+                        case DataSourceFormat.psv:
+                        case DataSourceFormat.scsv:
+                        case DataSourceFormat.sohsv:
+                        case DataSourceFormat.tsv:
+                        case DataSourceFormat.tsve:
+                            mapping = schemaRes.TableScheme.ToCsvMapping(detectionMode);
                             break;
                     }
                 }