Forum Discussion

JonBFabric's avatar
JonBFabric
Helper I
8 months ago
Solved

Lakehouse Table Generate Create Table

Hi,   I maintain server lakehouses, and due to issues with the deployment pipelines tend to apply schema changes through script, and to retain the data in the table use the following process:   ...
data engineering
Lakehouse
  • JonBFabric's avatar
    JonBFabric
    8 months ago

    One final update on this.

     

    The logfile containing the schema is not necessarily the most recent, and there is not necessarily only one version of the schema. There is a schema associated with every modification made to the table structure, be that the original creation or subsequent alterations. Consequently, the logfile we need is the most recent with a schema.

     

    from delta.tables import DeltaTable
    import json
     
    # Note that the table name must be lowercase
    table_path = '<Path_To_Table>'
     
    # Identify log files
    log_dir = f"{table_path}/_delta_log"
    files = [f.path for f in notebookutils.fs.ls(log_dir) if f.name.endswith(".json")]
     
    # Identify log files with a schema
    filesWithSchema = []

    for file in sorted(files, reverse=True) :
        content = notebookutils.fs.head(file, 5000000)
        JSONdocs = content.split('\n')

        for doc in JSONdocs:
            if 'schemaString' in doc:
                filesWithSchema.append(file)
     
     
    # Load the header for the latest log file containing a schema
    latest = sorted(filesWithSchema, reverse=True)[0]
    content = notebookutils.fs.head(latest, 5000000)
     
    # Extract the schema
    JSONdocs = content.split('\n')

    for doc in JSONdocs:
        if 'schemaString' in doc:
            schemaString = json.loads(doc).get("metaData", {}).get("schemaString")

    # Extract Field Metadata
    FieldList = []
    OrdinalPosition = 0

    for field in json.loads(schemaString).get("fields") :
        OrdinalPosition += 1
        FieldDetails = {}
        FieldDetails['FieldName'] = field.get("name")
        FieldDetails['Nullable'] = field.get("nullable")
        FieldDetails['OrdinalPosition'] = OrdinalPosition

        match field.get("type").split('(')[0]:
            case 'string':
                FieldDetails['SQLType'] = field.get("metadata").get("__CHAR_VARCHAR_TYPE_STRING")
            case 'timestamp':
                FieldDetails['SQLType'] = 'timestamp'
            case 'date':
                FieldDetails['SQLType'] = 'date'
            case 'integer':
                FieldDetails['SQLType'] = 'int'
            case 'short':
                FieldDetails['SQLType'] = 'smallint'
            case 'long':
                FieldDetails['SQLType'] = 'bigint'
            case 'decimal':
                FieldDetails['SQLType'] = field.get("type")
            case 'boolean':
                FieldDetails['SQLType'] = 'boolean'

        FieldList.append(FieldDetails)


    display(FieldList)
v-hashadapu's avatar
v-hashadapu
Community Support
8 months ago

Hi JonBFabric , Thank you for reaching out to the Microsoft Community Forum.

 

If Fabric is blocking SHOW CREATE TABLE and DESCRIBE TABLE EXTENDED only shows string, the next step is to read the Delta metadata directly through Spark, because the enforcement you are seeing (DELTA_EXCEED_CHAR_VARCHAR_LIMIT) comes from the schema stored in the Delta transaction log, not from the SQL endpoint. The length constraint is kept in the Delta log under metaData.schemaString and Spark/Delta will surface it correctly when you query the table through the Delta APIs. The simplest approach is to use a Spark notebook and load the table with DeltaTable.forPath(...).toDF(), which will show VarcharType(n) in the schema if the table was created with varchar(n). If that surface is not available, you can read the latest commit JSON in the _delta_log folder and print the metaData.schemaString field; that text contains the exact schema Spark is enforcing, including declared lengths.

 

Spark example you can run in a Fabric notebook to retrieve the metadata responsible for the enforcement:

from delta.tables import DeltaTable import json
table_path = "/lakehouses/<your-lakehouse>/Tables/<your-table>" # update this
dt = DeltaTable.forPath(spark, table_path) print(dt.toDF().schema) # shows VarcharType(n) if declared
log_dir = f"{table_path}/_delta_log" files = [f.path for f in dbutils.fs.ls(log_dir) if f.name.endswith(".json")] latest = sorted(files)[-1]
content = dbutils.fs.head(latest, 500000) commit = json.loads(content) print(commit.get("metaData", {}).get("schemaString"))

 

This will show you the exact schema stored in Delta and the constraint that triggers the length violation error. If the table is large and uses checkpoint parquet files, the same field appears in the checkpoint’s metaData struct. In short, the SQL endpoint cannot return the declared widths, but Spark and the Delta log always can.

 

Explore the lakehouse data with a notebook - Microsoft Fabric | Microsoft Learn

Data Types in Fabric Data Warehouse - Microsoft Fabric | Microsoft Learn

Delta Lake Logs in Warehouse - Microsoft Fabric | Microsoft Learn

What is a lakehouse? - Microsoft Fabric | Microsoft Learn

JonBFabric's avatar
JonBFabric
Helper I
8 months ago

Thanks. 

 

This didn't quite work out of the box, possibly because it was originally written for DataBricks rather than Fabric, but I have got it working. I will explan the differences as I go:

 

from delta.tables import DeltaTable
import json
table_path = "<Path_To_Table>" # update this
dt = DeltaTable.forPath(spark, table_path)
print(dt.toDF().schema) # shows VarcharType(n) if declared
 
.schema shows only string as type, so the last 2 lines can be removed.
 
log_dir = f"{table_path}/_delta_log"
files = [f.path for f in notebookutils.fs.ls(log_dir) if f.name.endswith(".json")]
latest = sorted(files)[-1]
content = notebookutils.fs.head(latest, 5000000)
 
Note that dbutils is now replaced by notebookutils.
 
content can not however be read as JSON, as it is infact 3 JSON documents seperated by 
'\n', and not just 1, and the document which contains the field metadata is the 2nd. 
 
schemaString = json.loads(content.split('\n')[1]).get("metaData", {}).get("schemaString")
 
schemaString is actually an embedded JSON document held as a string, and so has to be converted also. The following snippet then prints the field name and the actual sql type.
 
for field in json.loads(schemaString).get("fields"):
    print("FieldName:", field.get("name"), ", Type:", field.get("metadata").get("__CHAR_VARCHAR_TYPE_STRING"))
 
Certainly not a finished article, but it gives me what I need to build around.
 
Thanks for your help and patience. 
  • v-hashadapu's avatar
    v-hashadapu
    Community Support
    8 months ago

    Hi JonBFabric , Thanks for the update and the insights on how to solve this issue. We really appreciate it.

    If you have any queries, please feel free to create a new post, we are always happy to help.

  • JonBFabric's avatar
    JonBFabric
    Helper I
    8 months ago

    One final update on this.

     

    The logfile containing the schema is not necessarily the most recent, and there is not necessarily only one version of the schema. There is a schema associated with every modification made to the table structure, be that the original creation or subsequent alterations. Consequently, the logfile we need is the most recent with a schema.

     

    from delta.tables import DeltaTable
    import json
     
    # Note that the table name must be lowercase
    table_path = '<Path_To_Table>'
     
    # Identify log files
    log_dir = f"{table_path}/_delta_log"
    files = [f.path for f in notebookutils.fs.ls(log_dir) if f.name.endswith(".json")]
     
    # Identify log files with a schema
    filesWithSchema = []

    for file in sorted(files, reverse=True) :
        content = notebookutils.fs.head(file, 5000000)
        JSONdocs = content.split('\n')

        for doc in JSONdocs:
            if 'schemaString' in doc:
                filesWithSchema.append(file)
     
     
    # Load the header for the latest log file containing a schema
    latest = sorted(filesWithSchema, reverse=True)[0]
    content = notebookutils.fs.head(latest, 5000000)
     
    # Extract the schema
    JSONdocs = content.split('\n')

    for doc in JSONdocs:
        if 'schemaString' in doc:
            schemaString = json.loads(doc).get("metaData", {}).get("schemaString")

    # Extract Field Metadata
    FieldList = []
    OrdinalPosition = 0

    for field in json.loads(schemaString).get("fields") :
        OrdinalPosition += 1
        FieldDetails = {}
        FieldDetails['FieldName'] = field.get("name")
        FieldDetails['Nullable'] = field.get("nullable")
        FieldDetails['OrdinalPosition'] = OrdinalPosition

        match field.get("type").split('(')[0]:
            case 'string':
                FieldDetails['SQLType'] = field.get("metadata").get("__CHAR_VARCHAR_TYPE_STRING")
            case 'timestamp':
                FieldDetails['SQLType'] = 'timestamp'
            case 'date':
                FieldDetails['SQLType'] = 'date'
            case 'integer':
                FieldDetails['SQLType'] = 'int'
            case 'short':
                FieldDetails['SQLType'] = 'smallint'
            case 'long':
                FieldDetails['SQLType'] = 'bigint'
            case 'decimal':
                FieldDetails['SQLType'] = field.get("type")
            case 'boolean':
                FieldDetails['SQLType'] = 'boolean'

        FieldList.append(FieldDetails)


    display(FieldList)