Repository: chanzuckerberg/terraform-provider-snowflake
Pull request #577 (2021-06-16 00:04:02)

feat: Add file format resource
## Test Plan
* [ ] acceptance tests
* [ ] unit tests


wookasz: This is not my work. This is https://github.com/chanzuckerberg/terraform-provider-snowflake/pull/392 with merge conflicts resolved. Per https://github.com/chanzuckerberg/terraform-provider-snowflake/pull/392#issuecomment-857815429 just fixing up the merge conflict as requested in https://github.com/chanzuckerberg/terraform-provider-snowflake/pull/392#issuecomment-809622416.

I can also make review changes to try to get this pushed through.
alldoami: Can you run `make docs`?
alldoami: /ok-to-test sha=2223499
github-actions[bot]: Integration tests failure for [2223499](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/944459558)
wookasz: I extended the CSV file format acceptance test to cover all possible parameters, will add more but was hoping to get a run-through of the tests first as an easier way to make sure I'm formatting the verification strings correctly
alldoami: /ok-to-test sha=b529ba6
github-actions[bot]: Integration tests failure for [b529ba6](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/947508911)
wookasz: @alldoami added acceptance tests for all format types and fixed the slew of issues that this brought up. Great call on suggesting that.
alldoami: /ok-to-test sha=063ef8a
github-actions[bot]: Integration tests success for [063ef8a](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/957843414)

Patch:
diff --git a/docs/resources/file_format.md b/docs/resources/file_format.md
new file mode 100644
index 00000000..182eed8e
--- /dev/null
+++ b/docs/resources/file_format.md
@@ -0,0 +1,77 @@
+---
+# generated by https://github.com/hashicorp/terraform-plugin-docs
+page_title: "snowflake_file_format Resource - terraform-provider-snowflake"
+subcategory: ""
+description: |-
+  
+---
+
+# snowflake_file_format (Resource)
+
+
+
+## Example Usage
+
+```terraform
+resource "snowflake_file_format" "example_file_format" {
+  name        = "EXAMPLE_FILE_FORMAT"
+  database    = "EXAMPLE_DB"
+  schema      = "EXAMPLE_SCHEMA"
+  format_type = "CSV"
+}
+```
+
+<!-- schema generated by tfplugindocs -->
+## Schema
+
+### Required
+
+- **database** (String) The database in which to create the file format.
+- **format_type** (String) Specifies the format of the input files (for data loading) or output files (for data unloading).
+- **name** (String) Specifies the identifier for the file format; must be unique for the database and schema in which the file format is created.
+- **schema** (String) The schema in which to create the file format.
+
+### Optional
+
+- **allow_duplicate** (Boolean) Boolean that specifies to allow duplicate object field names (only the last one will be preserved).
+- **binary_as_text** (Boolean) Boolean that specifies whether to interpret columns with no defined logical data type as UTF-8 text.
+- **binary_format** (String) Defines the encoding format for binary input or output.
+- **comment** (String) Specifies a comment for the file format.
+- **compression** (String) Specifies the current compression algorithm for the data file.
+- **date_format** (String) Defines the format of date values in the data files (data loading) or table (data unloading).
+- **disable_auto_convert** (Boolean) Boolean that specifies whether the XML parser disables automatic conversion of numeric and Boolean values from text to native representation.
+- **disable_snowflake_data** (Boolean) Boolean that specifies whether the XML parser disables recognition of Snowflake semi-structured data tags.
+- **empty_field_as_null** (Boolean) Specifies whether to insert SQL NULL for empty fields in an input file, which are represented by two successive delimiters.
+- **enable_octal** (Boolean) Boolean that enables parsing of octal numbers.
+- **encoding** (String) String (constant) that specifies the character set of the source data when loading data into a table.
+- **error_on_column_count_mismatch** (Boolean) Boolean that specifies whether to generate a parsing error if the number of delimited columns (i.e. fields) in an input file does not match the number of columns in the corresponding table.
+- **escape** (String) Single character string used as the escape character for field values.
+- **escape_unenclosed_field** (String) Single character string used as the escape character for unenclosed field values only.
+- **field_delimiter** (String) Specifies one or more singlebyte or multibyte characters that separate fields in an input file (data loading) or unloaded file (data unloading).
+- **field_optionally_enclosed_by** (String) Character used to enclose strings.
+- **file_extension** (String) Specifies the extension for files unloaded to a stage.
+- **id** (String) The ID of this resource.
+- **ignore_utf8_errors** (Boolean) Boolean that specifies whether UTF-8 encoding errors produce error conditions.
+- **null_if** (List of String) String used to convert to and from SQL NULL.
+- **preserve_space** (Boolean) Boolean that specifies whether the XML parser preserves leading and trailing spaces in element content.
+- **record_delimiter** (String) Specifies one or more singlebyte or multibyte characters that separate records in an input file (data loading) or unloaded file (data unloading).
+- **replace_invalid_characters** (Boolean) Boolean that specifies whether to replace invalid UTF-8 characters with the Unicode replacement character (�).
+- **skip_blank_lines** (Boolean) Boolean that specifies to skip any blank lines encountered in the data files.
+- **skip_byte_order_mark** (Boolean) Boolean that specifies whether to skip the BOM (byte order mark), if present in a data file.
+- **skip_header** (Number) Number of lines at the start of the file to skip.
+- **strip_null_values** (Boolean) Boolean that instructs the JSON parser to remove object fields or array elements containing null values.
+- **strip_outer_array** (Boolean) Boolean that instructs the JSON parser to remove outer brackets.
+- **strip_outer_element** (Boolean) Boolean that specifies whether the XML parser strips out the outer XML element, exposing 2nd level elements as separate documents.
+- **time_format** (String) Defines the format of time values in the data files (data loading) or table (data unloading).
+- **timestamp_format** (String) Defines the format of timestamp values in the data files (data loading) or table (data unloading).
+- **trim_space** (Boolean) Boolean that specifies whether to remove white space from fields.
+- **validate_utf8** (Boolean) Boolean that specifies whether to validate UTF-8 character encoding in string column data.
+
+## Import
+
+Import is supported using the following syntax:
+
+```shell
+# format is database name | schema name | file format name
+terraform import snowflake_file_format.example 'dbName|schemaName|fileFormatName'
+```
diff --git a/examples/resources/snowflake_file_format/import.sh b/examples/resources/snowflake_file_format/import.sh
new file mode 100644
index 00000000..beb57975
--- /dev/null
+++ b/examples/resources/snowflake_file_format/import.sh
@@ -0,0 +1,2 @@
+# format is database name | schema name | file format name
+terraform import snowflake_file_format.example 'dbName|schemaName|fileFormatName'
diff --git a/examples/resources/snowflake_file_format/resource.tf b/examples/resources/snowflake_file_format/resource.tf
new file mode 100644
index 00000000..eae5b523
--- /dev/null
+++ b/examples/resources/snowflake_file_format/resource.tf
@@ -0,0 +1,6 @@
+resource "snowflake_file_format" "example_file_format" {
+  name        = "EXAMPLE_FILE_FORMAT"
+  database    = "EXAMPLE_DB"
+  schema      = "EXAMPLE_SCHEMA"
+  format_type = "CSV"
+}
diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go
index 089805dc..e46c737f 100644
--- a/pkg/provider/provider.go
+++ b/pkg/provider/provider.go
@@ -157,6 +157,7 @@ func getResources() map[string]*schema.Resource {
 		"snowflake_api_integration":           resources.APIIntegration(),
 		"snowflake_database":                  resources.Database(),
 		"snowflake_external_function":         resources.ExternalFunction(),
+		"snowflake_file_format":               resources.FileFormat(),
 		"snowflake_managed_account":           resources.ManagedAccount(),
 		"snowflake_masking_policy":            resources.MaskingPolicy(),
 		"snowflake_materialized_view":         resources.MaterializedView(),
diff --git a/pkg/resources/file_format.go b/pkg/resources/file_format.go
new file mode 100644
index 00000000..0431f4ae
--- /dev/null
+++ b/pkg/resources/file_format.go
@@ -0,0 +1,1143 @@
+package resources
+
+import (
+	"bytes"
+	"database/sql"
+	"encoding/csv"
+	"fmt"
+	"strings"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"
+	"github.com/pkg/errors"
+
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/snowflake"
+)
+
+const (
+	fileFormatIDDelimiter = '|'
+)
+
+// valid format type options for each File Format Type
+// https://docs.snowflake.com/en/sql-reference/sql/create-file-format.html#syntax
+var formatTypeOptions = map[string][]string{
+	"CSV": {
+		"compression",
+		"record_delimiter",
+		"field_delimiter",
+		"file_extension",
+		"skip_header",
+		"skip_blank_lines",
+		"date_format",
+		"time_format",
+		"timestamp_format",
+		"binary_format",
+		"escape",
+		"escape_unenclosed_field",
+		"trim_space",
+		"field_optionally_enclosed_by",
+		"null_if",
+		"error_on_column_count_mismatch",
+		"replace_invalid_characters",
+		"validate_utf8",
+		"empty_field_as_null",
+		"skip_byte_order_mark",
+		"encoding",
+	},
+	"JSON": {
+		"compression",
+		"date_format",
+		"time_format",
+		"timestamp_format",
+		"binary_format",
+		"trim_space",
+		"null_if",
+		"file_extension",
+		"enable_octal",
+		"allow_duplicate",
+		"strip_outer_array",
+		"strip_null_values",
+		"replace_invalid_characters",
+		"ignore_utf8_errors",
+		"skip_byte_order_mark",
+	},
+	"AVRO": {
+		"compression",
+		"trim_space",
+		"null_if",
+	},
+	"ORC": {
+		"trim_space",
+		"null_if",
+	},
+	"PARQUET": {
+		"compression",
+		"binary_as_text",
+		"trim_space",
+		"null_if",
+	},
+	"XML": {
+		"compression",
+		"ignore_utf8_errors",
+		"preserve_space",
+		"strip_outer_element",
+		"disable_snowflake_data",
+		"disable_auto_convert",
+		"skip_byte_order_mark",
+	},
+}
+
+var fileFormatSchema = map[string]*schema.Schema{
+	"name": {
+		Type:        schema.TypeString,
+		Required:    true,
+		Description: "Specifies the identifier for the file format; must be unique for the database and schema in which the file format is created.",
+		ForceNew:    true,
+	},
+	"database": {
+		Type:        schema.TypeString,
+		Required:    true,
+		Description: "The database in which to create the file format.",
+		ForceNew:    true,
+	},
+	"schema": {
+		Type:        schema.TypeString,
+		Required:    true,
+		Description: "The schema in which to create the file format.",
+		ForceNew:    true,
+	},
+	"format_type": {
+		Type:         schema.TypeString,
+		Required:     true,
+		Description:  "Specifies the format of the input files (for data loading) or output files (for data unloading).",
+		ForceNew:     true,
+		ValidateFunc: validation.StringInSlice([]string{"CSV", "JSON", "AVRO", "ORC", "PARQUET", "XML"}, true),
+	},
+	"compression": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Specifies the current compression algorithm for the data file.",
+	},
+	"record_delimiter": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Specifies one or more singlebyte or multibyte characters that separate records in an input file (data loading) or unloaded file (data unloading).",
+	},
+	"field_delimiter": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Specifies one or more singlebyte or multibyte characters that separate fields in an input file (data loading) or unloaded file (data unloading).",
+	},
+	"file_extension": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Specifies the extension for files unloaded to a stage.",
+	},
+	"skip_header": {
+		Type:        schema.TypeInt,
+		Optional:    true,
+		Description: "Number of lines at the start of the file to skip.",
+	},
+	"skip_blank_lines": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies to skip any blank lines encountered in the data files.",
+	},
+	"date_format": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Defines the format of date values in the data files (data loading) or table (data unloading).",
+	},
+	"time_format": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Defines the format of time values in the data files (data loading) or table (data unloading).",
+	},
+	"timestamp_format": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Defines the format of timestamp values in the data files (data loading) or table (data unloading).",
+	},
+	"binary_format": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Defines the encoding format for binary input or output.",
+	},
+	"escape": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Single character string used as the escape character for field values.",
+	},
+	"escape_unenclosed_field": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Single character string used as the escape character for unenclosed field values only.",
+	},
+	"trim_space": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies whether to remove white space from fields.",
+	},
+	"field_optionally_enclosed_by": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Character used to enclose strings.",
+	},
+	"null_if": {
+		Type:        schema.TypeList,
+		Elem:        &schema.Schema{Type: schema.TypeString},
+		Optional:    true,
+		Description: "String used to convert to and from SQL NULL.",
+	},
+	"error_on_column_count_mismatch": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies whether to generate a parsing error if the number of delimited columns (i.e. fields) in an input file does not match the number of columns in the corresponding table.",
+	},
+	"replace_invalid_characters": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies whether to replace invalid UTF-8 characters with the Unicode replacement character (�).",
+	},
+	"validate_utf8": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies whether to validate UTF-8 character encoding in string column data.",
+	},
+	"empty_field_as_null": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Specifies whether to insert SQL NULL for empty fields in an input file, which are represented by two successive delimiters.",
+	},
+	"skip_byte_order_mark": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies whether to skip the BOM (byte order mark), if present in a data file.",
+	},
+	"encoding": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "String (constant) that specifies the character set of the source data when loading data into a table.",
+	},
+	"enable_octal": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that enables parsing of octal numbers.",
+	},
+	"allow_duplicate": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies to allow duplicate object field names (only the last one will be preserved).",
+	},
+	"strip_outer_array": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that instructs the JSON parser to remove outer brackets.",
+	},
+	"strip_null_values": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that instructs the JSON parser to remove object fields or array elements containing null values.",
+	},
+	"ignore_utf8_errors": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies whether UTF-8 encoding errors produce error conditions.",
+	},
+	"binary_as_text": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies whether to interpret columns with no defined logical data type as UTF-8 text.",
+	},
+	"preserve_space": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies whether the XML parser preserves leading and trailing spaces in element content.",
+	},
+	"strip_outer_element": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies whether the XML parser strips out the outer XML element, exposing 2nd level elements as separate documents.",
+	},
+	"disable_snowflake_data": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies whether the XML parser disables recognition of Snowflake semi-structured data tags.",
+	},
+	"disable_auto_convert": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "Boolean that specifies whether the XML parser disables automatic conversion of numeric and Boolean values from text to native representation.",
+	},
+	"comment": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Specifies a comment for the file format.",
+	},
+}
+
+type fileFormatID struct {
+	DatabaseName   string
+	SchemaName     string
+	FileFormatName string
+}
+
+func (ffi *fileFormatID) String() (string, error) {
+	var buf bytes.Buffer
+	csvWriter := csv.NewWriter(&buf)
+	csvWriter.Comma = fileFormatIDDelimiter
+	err := csvWriter.WriteAll([][]string{{ffi.DatabaseName, ffi.SchemaName, ffi.FileFormatName}})
+	if err != nil {
+		return "", err
+	}
+
+	return strings.TrimSpace(buf.String()), nil
+}
+
+// FileFormat returns a pointer to the resource representing a file format
+func FileFormat() *schema.Resource {
+	return &schema.Resource{
+		Create: CreateFileFormat,
+		Read:   ReadFileFormat,
+		Update: UpdateFileFormat,
+		Delete: DeleteFileFormat,
+		Exists: FileFormatExists,
+
+		Schema: fileFormatSchema,
+		Importer: &schema.ResourceImporter{
+			State: schema.ImportStatePassthrough,
+		},
+	}
+}
+
+// CreateFileFormat implements schema.CreateFunc
+func CreateFileFormat(data *schema.ResourceData, meta interface{}) error {
+	db := meta.(*sql.DB)
+
+	dbName := data.Get("database").(string)
+	schemaName := data.Get("schema").(string)
+	fileFormatName := data.Get("name").(string)
+
+	builder := snowflake.FileFormat(fileFormatName, dbName, schemaName)
+
+	formatType := data.Get("format_type").(string)
+	builder.WithFormatType(formatType)
+
+	// Set optionals
+	if v, ok, err := getFormatTypeOption(data, formatType, "compression"); ok && err == nil {
+		builder.WithCompression(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "record_delimiter"); ok && err == nil {
+		builder.WithRecordDelimiter(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "field_delimiter"); ok && err == nil {
+		builder.WithFieldDelimiter(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "file_extension"); ok && err == nil {
+		builder.WithFileExtension(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "skip_header"); ok && err == nil {
+		builder.WithSkipHeader(v.(int))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "skip_blank_lines"); ok && err == nil {
+		builder.WithSkipBlankLines(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "date_format"); ok && err == nil {
+		builder.WithDateFormat(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "time_format"); ok && err == nil {
+		builder.WithTimeFormat(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "timestamp_format"); ok && err == nil {
+		builder.WithTimestampFormat(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "binary_format"); ok && err == nil {
+		builder.WithBinaryFormat(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "escape"); ok && err == nil {
+		builder.WithEscape(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "escape_unenclosed_field"); ok && err == nil {
+		builder.WithEscapeUnenclosedField(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "trim_space"); ok && err == nil {
+		builder.WithTrimSpace(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "field_optionally_enclosed_by"); ok && err == nil {
+		builder.WithFieldOptionallyEnclosedBy(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "null_if"); ok && err == nil {
+		builder.WithNullIf(expandStringList(v.([]interface{})))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "error_on_column_count_mismatch"); ok && err == nil {
+		builder.WithErrorOnColumnCountMismatch(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "replace_invalid_characters"); ok && err == nil {
+		builder.WithReplaceInvalidCharacters(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "validate_utf8"); ok && err == nil {
+		builder.WithValidateUTF8(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "empty_field_as_null"); ok && err == nil {
+		builder.WithEmptyFieldAsNull(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "skip_byte_order_mark"); ok && err == nil {
+		builder.WithSkipByteOrderMark(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "encoding"); ok && err == nil {
+		builder.WithEncoding(v.(string))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "enable_octal"); ok && err == nil {
+		builder.WithEnableOctal(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "allow_duplicate"); ok && err == nil {
+		builder.WithAllowDuplicate(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "strip_outer_array"); ok && err == nil {
+		builder.WithStripOuterArray(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "strip_null_values"); ok && err == nil {
+		builder.WithStripNullValues(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "ignore_utf8_errors"); ok && err == nil {
+		builder.WithIgnoreUTF8Errors(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "binary_as_text"); ok && err == nil {
+		builder.WithBinaryAsText(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "preserve_space"); ok && err == nil {
+		builder.WithPreserveSpace(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "strip_outer_element"); ok && err == nil {
+		builder.WithStripOuterElement(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "disable_snowflake_data"); ok && err == nil {
+		builder.WithDisableSnowflakeData(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok, err := getFormatTypeOption(data, formatType, "disable_auto_convert"); ok && err == nil {
+		builder.WithDisableAutoConvert(v.(bool))
+	} else if err != nil {
+		return err
+	}
+
+	if v, ok := data.GetOk("comment"); ok {
+		builder.WithComment(v.(string))
+	}
+
+	q := builder.Create()
+
+	err := snowflake.Exec(db, q)
+	if err != nil {
+		return errors.Wrapf(err, "error creating file format %v", fileFormatName)
+	}
+
+	fileFormatID := &fileFormatID{
+		DatabaseName:   dbName,
+		SchemaName:     schemaName,
+		FileFormatName: fileFormatName,
+	}
+	dataIDInput, err := fileFormatID.String()
+	if err != nil {
+		return err
+	}
+	data.SetId(dataIDInput)
+
+	return ReadFileFormat(data, meta)
+}
+
+// ReadFileFormat implements schema.ReadFunc
+func ReadFileFormat(data *schema.ResourceData, meta interface{}) error {
+	db := meta.(*sql.DB)
+	fileFormatID, err := fileFormatIDFromString(data.Id())
+	if err != nil {
+		return err
+	}
+
+	dbName := fileFormatID.DatabaseName
+	schemaName := fileFormatID.SchemaName
+	fileFormatName := fileFormatID.FileFormatName
+
+	ff := snowflake.FileFormat(fileFormatName, dbName, schemaName).Show()
+	row := snowflake.QueryRow(db, ff)
+
+	f, err := snowflake.ScanFileFormatShow(row)
+	if err != nil {
+		return err
+	}
+
+	opts, err := snowflake.ParseFormatOptions(f.FormatOptions.String)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("name", f.FileFormatName.String)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("database", f.DatabaseName.String)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("schema", f.SchemaName.String)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("format_type", opts.Type)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("compression", opts.Compression)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("record_delimiter", opts.RecordDelimiter)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("field_delimiter", opts.FieldDelimiter)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("file_extension", opts.FileExtension)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("skip_header", opts.SkipHeader)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("skip_blank_lines", opts.SkipBlankLines)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("date_format", opts.DateFormat)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("time_format", opts.TimeFormat)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("timestamp_format", opts.TimestampFormat)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("binary_format", opts.BinaryFormat)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("escape", opts.Escape)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("escape_unenclosed_field", opts.EscapeUnenclosedField)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("trim_space", opts.TrimSpace)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("field_optionally_enclosed_by", opts.FieldOptionallyEnclosedBy)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("null_if", opts.NullIf)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("error_on_column_count_mismatch", opts.ErrorOnColumnCountMismatch)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("replace_invalid_characters", opts.ReplaceInvalidCharacters)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("validate_utf8", opts.ValidateUTF8)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("empty_field_as_null", opts.EmptyFieldAsNull)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("skip_byte_order_mark", opts.SkipByteOrderMark)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("encoding", opts.Encoding)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("enable_octal", opts.EnabelOctal)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("allow_duplicate", opts.AllowDuplicate)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("strip_outer_array", opts.StripOuterArray)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("strip_null_values", opts.StripNullValues)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("ignore_utf8_errors", opts.IgnoreUTF8Errors)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("binary_as_text", opts.BinaryAsText)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("preserve_space", opts.PreserveSpace)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("strip_outer_element", opts.StripOuterElement)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("disable_snowflake_data", opts.DisableSnowflakeData)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("disable_auto_convert", opts.DisableAutoConvert)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("comment", f.Comment.String)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// UpdateFileFormat implements schema.UpdateFunc
+func UpdateFileFormat(data *schema.ResourceData, meta interface{}) error {
+	fileFormatID, err := fileFormatIDFromString(data.Id())
+	if err != nil {
+		return err
+	}
+
+	dbName := fileFormatID.DatabaseName
+	schemaName := fileFormatID.SchemaName
+	fileFormatName := fileFormatID.FileFormatName
+
+	builder := snowflake.FileFormat(fileFormatName, dbName, schemaName)
+	fmt.Println(builder)
+
+	db := meta.(*sql.DB)
+	if data.HasChange("compression") {
+		change := data.Get("compression")
+		q := builder.ChangeCompression(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format compression on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("record_delimiter") {
+		change := data.Get("record_delimiter")
+		q := builder.ChangeRecordDelimiter(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format record delimiter on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("field_delimiter") {
+		change := data.Get("field_delimiter")
+		q := builder.ChangeFieldDelimiter(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format field delimiter on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("file_extension") {
+		change := data.Get("file_extension")
+		q := builder.ChangeFileExtension(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format file extension on %v", data.Id())
+		}
+
+	}
+
+	if data.HasChange("skip_header") {
+		change := data.Get("skip_header")
+		q := builder.ChangeSkipHeader(change.(int))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format skip header on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("skip_blank_lines") {
+		change := data.Get("skip_blank_lines")
+		q := builder.ChangeSkipBlankLines(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format skip blank lines on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("date_format") {
+		change := data.Get("date_format")
+		q := builder.ChangeDateFormat(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format date format on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("time_format") {
+		change := data.Get("time_format")
+		q := builder.ChangeTimeFormat(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format time format on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("timestamp_format") {
+		change := data.Get("timestamp_format")
+		q := builder.ChangeTimestampFormat(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format timstamp format on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("binary_format") {
+		change := data.Get("binary_format")
+		q := builder.ChangeBinaryFormat(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format binary format on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("escape") {
+		change := data.Get("escape")
+		q := builder.ChangeEscape(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format escape on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("escape_unenclosed_field") {
+		change := data.Get("escape_unenclosed_field")
+		q := builder.ChangeEscapeUnenclosedField(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format escape_unenclosed_field on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("field_optionally_enclosed_by") {
+		change := data.Get("field_optionally_enclosed_by")
+		q := builder.ChangeFieldOptionallyEnclosedBy(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format field_optionally_enclosed_by on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("encoding") {
+		change := data.Get("encoding")
+		q := builder.ChangeEncoding(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format encoding on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("comment") {
+		change := data.Get("comment")
+		q := builder.ChangeComment(change.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format comment on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("trim_space") {
+		change := data.Get("trim_space")
+		q := builder.ChangeTrimSpace(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format trim_space on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("error_on_column_count_mismatch") {
+		change := data.Get("error_on_column_count_mismatch")
+		q := builder.ChangeErrorOnColumnCountMismatch(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format error_on_column_count_mismatch on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("replace_invalid_characters") {
+		change := data.Get("replace_invalid_characters")
+		q := builder.ChangeReplaceInvalidCharacters(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format replace_invalid_characters on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("validate_utf8") {
+		change := data.Get("validate_utf8")
+		q := builder.ChangeValidateUTF8(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format validate_utf8 on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("empty_field_as_null") {
+		change := data.Get("empty_field_as_null")
+		q := builder.ChangeEmptyFieldAsNull(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format empty_field_as_null on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("skip_byte_order_mark") {
+		change := data.Get("skip_byte_order_mark")
+		q := builder.ChangeSkipByteOrderMark(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format skip_byte_order_mark on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("enable_octal") {
+		change := data.Get("enable_octal")
+		q := builder.ChangeEnableOctal(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format enable_octal on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("allow_duplicate") {
+		change := data.Get("allow_duplicate")
+		q := builder.ChangeAllowDuplicate(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format allow_duplicate on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("strip_outer_array") {
+		change := data.Get("strip_outer_array")
+		q := builder.ChangeStripOuterArray(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format strip_outer_array on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("strip_null_values") {
+		change := data.Get("strip_null_values")
+		q := builder.ChangeStripNullValues(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format strip_null_values on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("ignore_utf8_errors") {
+		change := data.Get("ignore_utf8_errors")
+		q := builder.ChangeIgnoreUTF8Errors(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format ignore_utf8_errors on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("binary_as_text") {
+		change := data.Get("binary_as_text")
+		q := builder.ChangeBinaryAsText(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format binary_as_text on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("preserve_space") {
+		change := data.Get("preserve_space")
+		q := builder.ChangePreserveSpace(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format preserve_space on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("strip_outer_element") {
+		change := data.Get("strip_outer_element")
+		q := builder.ChangeStripOuterElement(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format strip_outer_element on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("disable_snowflake_data") {
+		change := data.Get("disable_snowflake_data")
+		q := builder.ChangeDisableSnowflakeData(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format disable_snowflake_data on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("disable_auto_convert") {
+		change := data.Get("disable_auto_convert")
+		q := builder.ChangeDisableAutoConvert(change.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format disable_auto_convert on %v", data.Id())
+		}
+	}
+
+	if data.HasChange("null_if") {
+		change := data.Get("null_if")
+		q := builder.ChangeNullIf(expandStringList(change.([]interface{})))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating file format null_if on %v", data.Id())
+		}
+	}
+
+	return ReadFileFormat(data, meta)
+}
+
+// DeleteFileFormat implements schema.DeleteFunc
+func DeleteFileFormat(data *schema.ResourceData, meta interface{}) error {
+	db := meta.(*sql.DB)
+	fileFormatID, err := fileFormatIDFromString(data.Id())
+	if err != nil {
+		return err
+	}
+
+	dbName := fileFormatID.DatabaseName
+	schemaName := fileFormatID.SchemaName
+	fileFormatName := fileFormatID.FileFormatName
+
+	q := snowflake.FileFormat(fileFormatName, dbName, schemaName).Drop()
+
+	err = snowflake.Exec(db, q)
+	if err != nil {
+		return errors.Wrapf(err, "error deleting file format %v", data.Id())
+	}
+
+	data.SetId("")
+
+	return nil
+}
+
+// FileFormatExists implements schema.ExistsFunc
+func FileFormatExists(data *schema.ResourceData, meta interface{}) (bool, error) {
+	db := meta.(*sql.DB)
+	fileFormatID, err := fileFormatIDFromString(data.Id())
+	if err != nil {
+		return false, err
+	}
+
+	dbName := fileFormatID.DatabaseName
+	schemaName := fileFormatID.SchemaName
+	fileFormatName := fileFormatID.FileFormatName
+
+	q := snowflake.FileFormat(fileFormatName, dbName, schemaName).Describe()
+	rows, err := db.Query(q)
+	if err != nil {
+		return false, err
+	}
+	defer rows.Close()
+
+	if rows.Next() {
+		return true, nil
+	}
+
+	return false, nil
+}
+
+func fileFormatIDFromString(stringID string) (*fileFormatID, error) {
+	reader := csv.NewReader(strings.NewReader(stringID))
+	reader.Comma = fileFormatIDDelimiter
+	lines, err := reader.ReadAll()
+	if err != nil {
+		return nil, fmt.Errorf("Not CSV compatible")
+	}
+
+	if len(lines) != 1 {
+		return nil, fmt.Errorf("1 line at a time")
+	}
+	if len(lines[0]) != 3 {
+		return nil, fmt.Errorf("4 fields allowed")
+	}
+
+	return &fileFormatID{
+		DatabaseName:   lines[0][0],
+		SchemaName:     lines[0][1],
+		FileFormatName: lines[0][2],
+	}, nil
+}
+
+func getFormatTypeOption(d *schema.ResourceData, formatType, formatTypeOption string) (interface{}, bool, error) {
+	validFormatTypeOptions := formatTypeOptions[formatType]
+	if v, ok := d.GetOk(formatTypeOption); ok {
+		if err := validateFormatTypeOptions(formatType, formatTypeOption, validFormatTypeOptions); err != nil {
+			return nil, true, err
+		}
+		return v, true, nil
+	}
+	return nil, false, nil
+}
+
+func validateFormatTypeOptions(formatType, formatTypeOption string, validFormatTypeOptions []string) error {
+	for _, f := range validFormatTypeOptions {
+		if f == formatTypeOption {
+			return nil
+		}
+	}
+	return fmt.Errorf("%v is an invalid format type option for format type %v", formatTypeOption, formatType)
+}
diff --git a/pkg/snowflake/file_format.go b/pkg/snowflake/file_format.go
new file mode 100644
index 00000000..937967cd
--- /dev/null
+++ b/pkg/snowflake/file_format.go
@@ -0,0 +1,623 @@
+package snowflake
+
+import (
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/jmoiron/sqlx"
+)
+
+// FileFormatBuilder abstracts the creation of SQL queries for a Snowflake file format
+type FileFormatBuilder struct {
+	name                       string
+	db                         string
+	schema                     string
+	formatType                 string
+	compression                string
+	recordDelimiter            string
+	fieldDelimiter             string
+	fileExtension              string
+	skipHeader                 int
+	skipBlankLines             bool
+	dateFormat                 string
+	timeFormat                 string
+	timestampFormat            string
+	binaryFormat               string
+	escape                     string
+	escapeUnenclosedField      string
+	trimSpace                  bool
+	fieldOptionallyEnclosedBy  string
+	nullIf                     []string
+	errorOnColumnCountMismatch bool
+	replaceInvalidCharacters   bool
+	validateUTF8               bool
+	emptyFieldAsNull           bool
+	skipByteOrderMark          bool
+	encoding                   string
+	enableOctal                bool
+	allowDuplicate             bool
+	stripOuterArray            bool
+	stripNullValues            bool
+	ignoreUTF8Errors           bool
+	binaryAsText               bool
+	preserveSpace              bool
+	stripOuterElement          bool
+	disableSnowflakeData       bool
+	disableAutoConvert         bool
+	comment                    string
+}
+
+// QualifiedName prepends the db and schema and escapes everything nicely
+func (ffb *FileFormatBuilder) QualifiedName() string {
+	var n strings.Builder
+
+	n.WriteString(fmt.Sprintf(`"%v"."%v"."%v"`, ffb.db, ffb.schema, ffb.name))
+
+	return n.String()
+}
+
+// WithFormatType adds a comment to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithFormatType(f string) *FileFormatBuilder {
+	ffb.formatType = f
+	return ffb
+}
+
+// WithCompression adds compression to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithCompression(c string) *FileFormatBuilder {
+	ffb.compression = c
+	return ffb
+}
+
+// WithRecordDelimiter adds a record delimiter to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithRecordDelimiter(r string) *FileFormatBuilder {
+	ffb.recordDelimiter = r
+	return ffb
+}
+
+// WithFieldDelimiter adds a field delimiter to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithFieldDelimiter(f string) *FileFormatBuilder {
+	ffb.fieldDelimiter = f
+	return ffb
+}
+
+// WithFileExtension adds a file extension to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithFileExtension(f string) *FileFormatBuilder {
+	ffb.fileExtension = f
+	return ffb
+}
+
+// WithSkipHeader adds skip header to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithSkipHeader(n int) *FileFormatBuilder {
+	ffb.skipHeader = n
+	return ffb
+}
+
+// WithSkipBlankLines adds skip blank lines to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithSkipBlankLines(n bool) *FileFormatBuilder {
+	ffb.skipBlankLines = n
+	return ffb
+}
+
+// WithDateFormat adds date format to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithDateFormat(s string) *FileFormatBuilder {
+	ffb.dateFormat = s
+	return ffb
+}
+
+// WithTimeFormat adds time format to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithTimeFormat(s string) *FileFormatBuilder {
+	ffb.timeFormat = s
+	return ffb
+}
+
+// WithTimestampFormat adds timestamp format to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithTimestampFormat(s string) *FileFormatBuilder {
+	ffb.timestampFormat = s
+	return ffb
+}
+
+// WithBinaryFormat adds binary format to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithBinaryFormat(s string) *FileFormatBuilder {
+	ffb.binaryFormat = s
+	return ffb
+}
+
+// WithEscape adds escape to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithEscape(s string) *FileFormatBuilder {
+	ffb.escape = s
+	return ffb
+}
+
+// WithEscapeUnenclosedField adds escape unenclosed field to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithEscapeUnenclosedField(s string) *FileFormatBuilder {
+	ffb.escapeUnenclosedField = s
+	return ffb
+}
+
+// WithTrimSpace adds trim space to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithTrimSpace(n bool) *FileFormatBuilder {
+	ffb.trimSpace = n
+	return ffb
+}
+
+// WithFieldOptionallyEnclosedBy adds field optionally enclosed by to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithFieldOptionallyEnclosedBy(s string) *FileFormatBuilder {
+	ffb.fieldOptionallyEnclosedBy = s
+	return ffb
+}
+
+// WithNullIf adds null if to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithNullIf(s []string) *FileFormatBuilder {
+	ffb.nullIf = s
+	return ffb
+}
+
+// WithErrorOnColumnCountMismatch adds error on column count mistmatch to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithErrorOnColumnCountMismatch(n bool) *FileFormatBuilder {
+	ffb.errorOnColumnCountMismatch = n
+	return ffb
+}
+
+// WithReplaceInvalidCharacters adds replace invalid characters to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithReplaceInvalidCharacters(n bool) *FileFormatBuilder {
+	ffb.replaceInvalidCharacters = n
+	return ffb
+}
+
+// WithValidateUTF8 adds validate utf8 to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithValidateUTF8(n bool) *FileFormatBuilder {
+	ffb.validateUTF8 = n
+	return ffb
+}
+
+// WithEmptyFieldAsNull adds empty field as null to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithEmptyFieldAsNull(n bool) *FileFormatBuilder {
+	ffb.emptyFieldAsNull = n
+	return ffb
+}
+
+// WithSkipByteOrderMark adds skip byte order mark to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithSkipByteOrderMark(n bool) *FileFormatBuilder {
+	ffb.skipByteOrderMark = n
+	return ffb
+}
+
+// WithEnableOctal adds enable octal to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithEnableOctal(n bool) *FileFormatBuilder {
+	ffb.enableOctal = n
+	return ffb
+}
+
+// WithAllowDuplicate adds allow duplicate to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithAllowDuplicate(n bool) *FileFormatBuilder {
+	ffb.allowDuplicate = n
+	return ffb
+}
+
+// WithStripOuterArray adds strip outer array to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithStripOuterArray(n bool) *FileFormatBuilder {
+	ffb.stripOuterArray = n
+	return ffb
+}
+
+// WithStripNullValues adds strip null values to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithStripNullValues(n bool) *FileFormatBuilder {
+	ffb.stripNullValues = n
+	return ffb
+}
+
+// WithIgnoreUTF8Errors adds ignore UTF8 errors to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithIgnoreUTF8Errors(n bool) *FileFormatBuilder {
+	ffb.ignoreUTF8Errors = n
+	return ffb
+}
+
+// WithBinaryAsText adds binary as text to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithBinaryAsText(n bool) *FileFormatBuilder {
+	ffb.binaryAsText = n
+	return ffb
+}
+
+// WithPreserveSpace adds preserve space to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithPreserveSpace(n bool) *FileFormatBuilder {
+	ffb.preserveSpace = n
+	return ffb
+}
+
+// WithStripOuterElement adds strip outer element to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithStripOuterElement(n bool) *FileFormatBuilder {
+	ffb.stripOuterElement = n
+	return ffb
+}
+
+// WithDisableSnowflakeData adds disable Snowflake data to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithDisableSnowflakeData(n bool) *FileFormatBuilder {
+	ffb.disableSnowflakeData = n
+	return ffb
+}
+
+// WithDisableAutoConvert adds disbale auto convert to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithDisableAutoConvert(n bool) *FileFormatBuilder {
+	ffb.disableAutoConvert = n
+	return ffb
+}
+
+// WithEncoding adds encoding to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithEncoding(e string) *FileFormatBuilder {
+	ffb.encoding = e
+	return ffb
+}
+
+// WithComment adds a comment to the FileFormatBuilder
+func (ffb *FileFormatBuilder) WithComment(c string) *FileFormatBuilder {
+	ffb.comment = c
+	return ffb
+}
+
+// FileFormat returns a pointer to a Builder that abstracts the DDL operations for a file format.
+//
+// Supported DDL operations are:
+//   - CREATE FILE FORMAT
+//   - ALTER FILE FORMAT
+//   - DROP FILE FORMAT
+//   - SHOW FILE FORMATS
+//   - DESCRIBE FILE FORMAT
+//
+// [Snowflake Reference](https://docs.snowflake.com/en/sql-reference/sql/create-file-format.html)
+func FileFormat(name, db, schema string) *FileFormatBuilder {
+	return &FileFormatBuilder{
+		name:   name,
+		db:     db,
+		schema: schema,
+	}
+}
+
+// Create returns the SQL query that will create a new file format.
+func (ffb *FileFormatBuilder) Create() string {
+	q := strings.Builder{}
+	q.WriteString(`CREATE`)
+
+	q.WriteString(fmt.Sprintf(` FILE FORMAT %v`, ffb.QualifiedName()))
+	q.WriteString(fmt.Sprintf(` TYPE = '%v'`, ffb.formatType))
+
+	if ffb.compression != "" {
+		q.WriteString(fmt.Sprintf(` COMPRESSION = '%v'`, ffb.compression))
+	}
+
+	if ffb.recordDelimiter != "" {
+		q.WriteString(fmt.Sprintf(` RECORD_DELIMITER = '%v'`, ffb.recordDelimiter))
+	}
+
+	if ffb.fieldDelimiter != "" {
+		q.WriteString(fmt.Sprintf(` FIELD_DELIMITER = '%v'`, ffb.fieldDelimiter))
+	}
+
+	if ffb.fileExtension != "" {
+		q.WriteString(fmt.Sprintf(` FILE_EXTENSION = '%v'`, ffb.fileExtension))
+	}
+
+	if ffb.skipHeader > 0 {
+		q.WriteString(fmt.Sprintf(` SKIP_HEADER = %v`, ffb.skipHeader))
+	}
+
+	if ffb.dateFormat != "" {
+		q.WriteString(fmt.Sprintf(` DATE_FORMAT = '%v'`, ffb.dateFormat))
+	}
+
+	if ffb.timeFormat != "" {
+		q.WriteString(fmt.Sprintf(` TIME_FORMAT = '%v'`, ffb.timeFormat))
+	}
+
+	if ffb.timestampFormat != "" {
+		q.WriteString(fmt.Sprintf(` TIMESTAMP_FORMAT = '%v'`, ffb.timestampFormat))
+	}
+
+	if ffb.binaryFormat != "" {
+		q.WriteString(fmt.Sprintf(` BINARY_FORMAT = '%v'`, ffb.binaryFormat))
+	}
+
+	if ffb.escape != "" {
+		q.WriteString(fmt.Sprintf(` ESCAPE = '%v'`, EscapeString(ffb.escape)))
+	}
+
+	if ffb.escapeUnenclosedField != "" {
+		q.WriteString(fmt.Sprintf(` ESCAPE_UNENCLOSED_FIELD = '%v'`, ffb.escapeUnenclosedField))
+	}
+
+	if ffb.fieldOptionallyEnclosedBy != "" {
+		q.WriteString(fmt.Sprintf(` FIELD_OPTIONALLY_ENCLOSED_BY = '%v'`, EscapeString(ffb.fieldOptionallyEnclosedBy)))
+	}
+
+	if len(ffb.nullIf) > 0 {
+		nullIfStr := "'" + strings.Join(ffb.nullIf, "', '") + "'"
+		q.WriteString(fmt.Sprintf(` NULL_IF = (%v)`, nullIfStr))
+	} else if strings.ToUpper(ffb.formatType) != "XML" {
+		q.WriteString(` NULL_IF = ()`)
+	}
+
+	if ffb.encoding != "" {
+		q.WriteString(fmt.Sprintf(` ENCODING = '%v'`, ffb.encoding))
+	}
+
+	// set boolean values
+	if ffb.formatType == "CSV" {
+		q.WriteString(fmt.Sprintf(` SKIP_BLANK_LINES = %v`, ffb.skipBlankLines))
+		q.WriteString(fmt.Sprintf(` TRIM_SPACE = %v`, ffb.trimSpace))
+		q.WriteString(fmt.Sprintf(` ERROR_ON_COLUMN_COUNT_MISMATCH = %v`, ffb.errorOnColumnCountMismatch))
+		q.WriteString(fmt.Sprintf(` REPLACE_INVALID_CHARACTERS = %v`, ffb.replaceInvalidCharacters))
+		q.WriteString(fmt.Sprintf(` VALIDATE_UTF8 = %v`, ffb.validateUTF8))
+		q.WriteString(fmt.Sprintf(` EMPTY_FIELD_AS_NULL = %v`, ffb.emptyFieldAsNull))
+		q.WriteString(fmt.Sprintf(` SKIP_BYTE_ORDER_MARK = %v`, ffb.skipByteOrderMark))
+	} else if ffb.formatType == "JSON" {
+		q.WriteString(fmt.Sprintf(` TRIM_SPACE = %v`, ffb.trimSpace))
+		q.WriteString(fmt.Sprintf(` ENABLE_OCTAL = %v`, ffb.enableOctal))
+		q.WriteString(fmt.Sprintf(` ALLOW_DUPLICATE = %v`, ffb.allowDuplicate))
+		q.WriteString(fmt.Sprintf(` STRIP_OUTER_ARRAY = %v`, ffb.stripOuterArray))
+		q.WriteString(fmt.Sprintf(` STRIP_NULL_VALUES = %v`, ffb.stripNullValues))
+		q.WriteString(fmt.Sprintf(` REPLACE_INVALID_CHARACTERS = %v`, ffb.replaceInvalidCharacters))
+		q.WriteString(fmt.Sprintf(` IGNORE_UTF8_ERRORS = %v`, ffb.ignoreUTF8Errors))
+		q.WriteString(fmt.Sprintf(` SKIP_BYTE_ORDER_MARK = %v`, ffb.skipByteOrderMark))
+	} else if ffb.formatType == "AVRO" || ffb.formatType == "ORC" {
+		q.WriteString(fmt.Sprintf(` TRIM_SPACE = %v`, ffb.trimSpace))
+	} else if ffb.formatType == "PARQUET" {
+		q.WriteString(fmt.Sprintf(` BINARY_AS_TEXT = %v`, ffb.binaryAsText))
+		q.WriteString(fmt.Sprintf(` TRIM_SPACE = %v`, ffb.trimSpace))
+	} else if ffb.formatType == "XML" {
+		q.WriteString(fmt.Sprintf(` IGNORE_UTF8_ERRORS = %v`, ffb.ignoreUTF8Errors))
+		q.WriteString(fmt.Sprintf(` PRESERVE_SPACE = %v`, ffb.preserveSpace))
+		q.WriteString(fmt.Sprintf(` STRIP_OUTER_ELEMENT = %v`, ffb.stripOuterElement))
+		q.WriteString(fmt.Sprintf(` DISABLE_SNOWFLAKE_DATA = %v`, ffb.disableSnowflakeData))
+		q.WriteString(fmt.Sprintf(` DISABLE_AUTO_CONVERT = %v`, ffb.disableAutoConvert))
+		q.WriteString(fmt.Sprintf(` SKIP_BYTE_ORDER_MARK = %v`, ffb.skipByteOrderMark))
+	}
+
+	if ffb.comment != "" {
+		q.WriteString(fmt.Sprintf(` COMMENT = '%v'`, EscapeString(ffb.comment)))
+	}
+
+	return q.String()
+}
+
+// ChangeComment returns the SQL query that will update the comment on the file format.
+func (ffb *FileFormatBuilder) ChangeComment(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET COMMENT = '%v'`, ffb.QualifiedName(), c)
+}
+
+// RemoveComment returns the SQL query that will remove the comment on the file format.
+func (ffb *FileFormatBuilder) RemoveComment() string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v UNSET COMMENT`, ffb.QualifiedName())
+}
+
+// ChangeCompression returns the SQL query that will update the compression on the file format.
+func (ffb *FileFormatBuilder) ChangeCompression(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET COMPRESSION = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeRecordDelimiter returns the SQL query that will update the record delimiter on the file format.
+func (ffb *FileFormatBuilder) ChangeRecordDelimiter(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET RECORD_DELIMITER = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeDateFormat returns the SQL query that will update the date format on the file format.
+func (ffb *FileFormatBuilder) ChangeDateFormat(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET DATE_FORMAT = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeTimeFormat returns the SQL query that will update the time format on the file format.
+func (ffb *FileFormatBuilder) ChangeTimeFormat(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET TIME_FORMAT = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeTimestampFormat returns the SQL query that will update the timestamp format on the file format.
+func (ffb *FileFormatBuilder) ChangeTimestampFormat(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET TIMESTAMP_FORMAT = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeBinaryFormat returns the SQL query that will update the binary format on the file format.
+func (ffb *FileFormatBuilder) ChangeBinaryFormat(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET BINARY_FORMAT = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeErrorOnColumnCountMismatch returns the SQL query that will update the error_on_column_count_mismatch on the file format.
+func (ffb *FileFormatBuilder) ChangeErrorOnColumnCountMismatch(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET ERROR_ON_COLUMN_COUNT_MISMATCH = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeValidateUTF8 returns the SQL query that will update the error_on_column_count_mismatch on the file format.
+func (ffb *FileFormatBuilder) ChangeValidateUTF8(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET VALIDATE_UTF8 = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeEmptyFieldAsNull returns the SQL query that will update the error_on_column_count_mismatch on the file format.
+func (ffb *FileFormatBuilder) ChangeEmptyFieldAsNull(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET EMPTY_FIELD_AS_NULL = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeEscape returns the SQL query that will update the escape on the file format.
+func (ffb *FileFormatBuilder) ChangeEscape(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET ESCAPE = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeEscapeUnenclosedField returns the SQL query that will update the escape unenclosed field on the file format.
+func (ffb *FileFormatBuilder) ChangeEscapeUnenclosedField(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET ESCAPE_UNENCLOSED_FIELD = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeFileExtension returns the SQL query that will update the FILE_EXTENSION on the file format.
+func (ffb *FileFormatBuilder) ChangeFileExtension(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET FILE_EXTENSION = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeFieldDelimiter returns the SQL query that will update the FIELD_DELIMITER on the file format.
+func (ffb *FileFormatBuilder) ChangeFieldDelimiter(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET FIELD_DELIMITER = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeFieldOptionallyEnclosedBy returns the SQL query that will update the field optionally enclosed by on the file format.
+func (ffb *FileFormatBuilder) ChangeFieldOptionallyEnclosedBy(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET FIELD_OPTIONALLY_ENCLOSED_BY = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeNullIf returns the SQL query that will update the null if on the file format.
+func (ffb *FileFormatBuilder) ChangeNullIf(c []string) string {
+	nullIfStr := ""
+	if len(c) > 0 {
+		nullIfStr = "'" + strings.Join(c, "', '") + "'"
+	}
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET NULL_IF = (%v)`, ffb.QualifiedName(), nullIfStr)
+}
+
+// ChangeEncoding returns the SQL query that will update the encoding on the file format.
+func (ffb *FileFormatBuilder) ChangeEncoding(c string) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET ENCODING = '%v'`, ffb.QualifiedName(), c)
+}
+
+// ChangeSkipHeader returns the SQL query that will update the skip header on the file format.
+func (ffb *FileFormatBuilder) ChangeSkipHeader(c int) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET SKIP_HEADER = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeSkipBlankLines returns the SQL query that will update SKIP_BLANK_LINES on the file format.
+func (ffb *FileFormatBuilder) ChangeSkipBlankLines(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET SKIP_BLANK_LINES = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeTrimSpace returns the SQL query that will update TRIM_SPACE on the file format.
+func (ffb *FileFormatBuilder) ChangeTrimSpace(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET TRIM_SPACE = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeEnableOctal returns the SQL query that will update ENABLE_OCTAL on the file format.
+func (ffb *FileFormatBuilder) ChangeEnableOctal(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET ENABLE_OCTAL = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeAllowDuplicate returns the SQL query that will update ALLOW_DUPLICATE on the file format.
+func (ffb *FileFormatBuilder) ChangeAllowDuplicate(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET ALLOW_DUPLICATE = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeStripOuterArray returns the SQL query that will update STRIP_OUTER_ARRAY on the file format.
+func (ffb *FileFormatBuilder) ChangeStripOuterArray(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET STRIP_OUTER_ARRAY = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeStripNullValues returns the SQL query that will update STRIP_NULL_VALUES on the file format.
+func (ffb *FileFormatBuilder) ChangeStripNullValues(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET STRIP_NULL_VALUES = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeReplaceInvalidCharacters returns the SQL query that will update REPLACE_INVALID_CHARACTERS on the file format.
+func (ffb *FileFormatBuilder) ChangeReplaceInvalidCharacters(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET REPLACE_INVALID_CHARACTERS = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeIgnoreUTF8Errors returns the SQL query that will update IGNORE_UTF8_ERRORS on the file format.
+func (ffb *FileFormatBuilder) ChangeIgnoreUTF8Errors(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET IGNORE_UTF8_ERRORS = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeSkipByteOrderMark returns the SQL query that will update SKIP_BYTE_ORDER_MARK on the file format.
+func (ffb *FileFormatBuilder) ChangeSkipByteOrderMark(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET SKIP_BYTE_ORDER_MARK = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeBinaryAsText returns the SQL query that will update BINARY_AS_TEXT on the file format.
+func (ffb *FileFormatBuilder) ChangeBinaryAsText(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET BINARY_AS_TEXT = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangePreserveSpace returns the SQL query that will update PRESERVE_SPACE on the file format.
+func (ffb *FileFormatBuilder) ChangePreserveSpace(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET PRESERVE_SPACE = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeStripOuterElement returns the SQL query that will update STRIP_OUTER_ELEMENT on the file format.
+func (ffb *FileFormatBuilder) ChangeStripOuterElement(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET STRIP_OUTER_ELEMENT = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeDisableSnowflakeData returns the SQL query that will update DISABLE_SNOWFLAKE_DATA on the file format.
+func (ffb *FileFormatBuilder) ChangeDisableSnowflakeData(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET DISABLE_SNOWFLAKE_DATA = %v`, ffb.QualifiedName(), c)
+}
+
+// ChangeDisableAutoConvert returns the SQL query that will update DISABLE_AUTO_CONVERT on the file format.
+func (ffb *FileFormatBuilder) ChangeDisableAutoConvert(c bool) string {
+	return fmt.Sprintf(`ALTER FILE FORMAT %v SET DISABLE_AUTO_CONVERT = %v`, ffb.QualifiedName(), c)
+}
+
+// Drop returns the SQL query that will drop a file format.
+func (ffb *FileFormatBuilder) Drop() string {
+	return fmt.Sprintf(`DROP FILE FORMAT %v`, ffb.QualifiedName())
+}
+
+// Describe returns the SQL query that will describe a file format..
+func (ffb *FileFormatBuilder) Describe() string {
+	return fmt.Sprintf(`DESCRIBE FILE FORMAT %v`, ffb.QualifiedName())
+}
+
+// Show returns the SQL query that will show a file format.
+func (ffb *FileFormatBuilder) Show() string {
+	return fmt.Sprintf(`SHOW FILE FORMATS LIKE '%v' IN DATABASE "%v"`, ffb.name, ffb.db)
+}
+
+type fileFormatShow struct {
+	CreatedOn      sql.NullString `db:"created_on"`
+	FileFormatName sql.NullString `db:"name"`
+	DatabaseName   sql.NullString `db:"database_name"`
+	SchemaName     sql.NullString `db:"schema_name"`
+	FormatType     sql.NullString `db:"type"`
+	Owner          sql.NullString `db:"owner"`
+	Comment        sql.NullString `db:"comment"`
+	FormatOptions  sql.NullString `db:"format_options"`
+}
+
+type fileFormatOptions struct {
+	Type                       string   `json:"TYPE"`
+	Compression                string   `json:"COMPRESSION,omitempty"`
+	RecordDelimiter            string   `json:"RECORD_DELIMITER,omitempty"`
+	FieldDelimiter             string   `json:"FIELD_DELIMITER,omitempty"`
+	FileExtension              string   `json:"FILE_EXTENSION,omitempty"`
+	SkipHeader                 int      `json:"SKIP_HEADER,omitempty"`
+	DateFormat                 string   `json:"DATE_FORMAT,omitempty"`
+	TimeFormat                 string   `json:"TIME_FORMAT,omitempty"`
+	TimestampFormat            string   `json:"TIMESTAMP_FORMAT,omitempty"`
+	BinaryFormat               string   `json:"BINARY_FORMAT,omitempty"`
+	Escape                     string   `json:"ESCAPE,omitempty"`
+	EscapeUnenclosedField      string   `json:"ESCAPE_UNENCLOSED_FIELD,omitempty"`
+	TrimSpace                  bool     `json:"TRIM_SPACE,omitempty"`
+	FieldOptionallyEnclosedBy  string   `json:"FIELD_OPTIONALLY_ENCLOSED_BY,omitempty"`
+	NullIf                     []string `json:"NULL_IF,omitempty"`
+	ErrorOnColumnCountMismatch bool     `json:"ERROR_ON_COLUMN_COUNT_MISMATCH,omitempty"`
+	ValidateUTF8               bool     `json:"VALIDATE_UTF8,omitempty"`
+	SkipBlankLines             bool     `json:"SKIP_BLANK_LINES,omitempty"`
+	ReplaceInvalidCharacters   bool     `json:"REPLACE_INVALID_CHARACTERS,omitempty"`
+	EmptyFieldAsNull           bool     `json:"EMPTY_FIELD_AS_NULL,omitempty"`
+	SkipByteOrderMark          bool     `json:"SKIP_BYTE_ORDER_MARK,omitempty"`
+	Encoding                   string   `json:"ENCODING,omitempty"`
+	EnabelOctal                bool     `json:"ENABLE_OCTAL,omitempty"`
+	AllowDuplicate             bool     `json:"ALLOW_DUPLICATE,omitempty"`
+	StripOuterArray            bool     `json:"STRIP_OUTER_ARRAY,omitempty"`
+	StripNullValues            bool     `json:"STRIP_NULL_VALUES,omitempty"`
+	IgnoreUTF8Errors           bool     `json:"IGNORE_UTF8_ERRORS,omitempty"`
+	BinaryAsText               bool     `json:"BINARY_AS_TEXT,omitempty"`
+	PreserveSpace              bool     `json:"PRESERVE_SPACE,omitempty"`
+	StripOuterElement          bool     `json:"STRIP_OUTER_ELEMENT,omitempty"`
+	DisableSnowflakeData       bool     `json:"DISABLE_SNOWFLAKE_DATA,omitempty"`
+	DisableAutoConvert         bool     `json:"DISABLE_AUTO_CONVERT,omitempty"`
+}
+
+func ScanFileFormatShow(row *sqlx.Row) (*fileFormatShow, error) {
+	r := &fileFormatShow{}
+	err := row.StructScan(r)
+	return r, err
+}
+
+func ParseFormatOptions(fileOptions string) (*fileFormatOptions, error) {
+	ff := &fileFormatOptions{}
+	err := json.Unmarshal([]byte(fileOptions), ff)
+	return ff, err
+}


Test patch:
diff --git a/pkg/resources/file_format_acceptance_test.go b/pkg/resources/file_format_acceptance_test.go
new file mode 100644
index 00000000..55ac17bb
--- /dev/null
+++ b/pkg/resources/file_format_acceptance_test.go
@@ -0,0 +1,372 @@
+package resources_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/acctest"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+)
+
+func TestAcc_FileFormatCSV(t *testing.T) {
+	accName := acctest.RandStringFromCharSet(10, acctest.CharSetAlpha)
+
+	resource.ParallelTest(t, resource.TestCase{
+		Providers: providers(),
+		Steps: []resource.TestStep{
+			{
+				Config: fileFormatConfigCSV(accName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "name", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "format_type", "CSV"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "compression", "GZIP"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "record_delimiter", "\r"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "field_delimiter", ";"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "file_extension", ".ssv"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "skip_header", "1"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "skip_blank_lines", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "date_format", "YYY-MM-DD"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "time_format", "HH24:MI"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "timestamp_format", "YYYY-MM-DD HH24:MI:SS.FFTZH:TZM"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "binary_format", "UTF8"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "escape", "\\"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "escape_unenclosed_field", "!"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "trim_space", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "field_optionally_enclosed_by", "'"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "null_if.#", "1"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "null_if.0", "NULL"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "error_on_column_count_mismatch", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "replace_invalid_characters", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "validate_utf8", "false"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "empty_field_as_null", "false"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "skip_byte_order_mark", "false"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "encoding", "UTF-16"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "comment", "Terraform acceptance test"),
+				),
+			},
+		},
+	})
+}
+
+func TestAcc_FileFormatJSON(t *testing.T) {
+	accName := acctest.RandStringFromCharSet(10, acctest.CharSetAlpha)
+
+	resource.ParallelTest(t, resource.TestCase{
+		Providers: providers(),
+		Steps: []resource.TestStep{
+			{
+				Config: fileFormatConfigJSON(accName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "name", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "format_type", "JSON"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "compression", "GZIP"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "date_format", "YYY-MM-DD"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "time_format", "HH24:MI"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "timestamp_format", "YYYY-MM-DD HH24:MI:SS.FFTZH:TZM"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "binary_format", "UTF8"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "trim_space", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "null_if.#", "1"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "null_if.0", "NULL"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "file_extension", ".jsn"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "enable_octal", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "allow_duplicate", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "strip_outer_array", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "strip_null_values", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "ignore_utf8_errors", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "skip_byte_order_mark", "false"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "comment", "Terraform acceptance test"),
+				),
+			},
+		},
+	})
+}
+
+func TestAcc_FileFormatAvro(t *testing.T) {
+	accName := acctest.RandStringFromCharSet(10, acctest.CharSetAlpha)
+
+	resource.ParallelTest(t, resource.TestCase{
+		Providers: providers(),
+		Steps: []resource.TestStep{
+			{
+				Config: fileFormatConfigAvro(accName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "name", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "format_type", "AVRO"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "compression", "GZIP"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "trim_space", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "null_if.#", "1"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "null_if.0", "NULL"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "comment", "Terraform acceptance test"),
+				),
+			},
+		},
+	})
+}
+
+func TestAcc_FileFormatORC(t *testing.T) {
+	accName := acctest.RandStringFromCharSet(10, acctest.CharSetAlpha)
+
+	resource.ParallelTest(t, resource.TestCase{
+		Providers: providers(),
+		Steps: []resource.TestStep{
+			{
+				Config: fileFormatConfigORC(accName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "name", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "format_type", "ORC"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "trim_space", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "null_if.#", "1"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "null_if.0", "NULL"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "comment", "Terraform acceptance test"),
+				),
+			},
+		},
+	})
+}
+
+func TestAcc_FileFormatParquet(t *testing.T) {
+	accName := acctest.RandStringFromCharSet(10, acctest.CharSetAlpha)
+
+	resource.ParallelTest(t, resource.TestCase{
+		Providers: providers(),
+		Steps: []resource.TestStep{
+			{
+				Config: fileFormatConfigParquet(accName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "name", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "format_type", "PARQUET"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "compression", "SNAPPY"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "binary_as_text", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "trim_space", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "null_if.#", "1"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "null_if.0", "NULL"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "comment", "Terraform acceptance test"),
+				),
+			},
+		},
+	})
+}
+
+func TestAcc_FileFormatXML(t *testing.T) {
+	accName := acctest.RandStringFromCharSet(10, acctest.CharSetAlpha)
+
+	resource.ParallelTest(t, resource.TestCase{
+		Providers: providers(),
+		Steps: []resource.TestStep{
+			{
+				Config: fileFormatConfigXML(accName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "name", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "format_type", "XML"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "compression", "GZIP"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "preserve_space", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "strip_outer_element", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "disable_snowflake_data", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "disable_auto_convert", "true"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "skip_byte_order_mark", "false"),
+					resource.TestCheckResourceAttr("snowflake_file_format.test", "comment", "Terraform acceptance test"),
+				),
+			},
+		},
+	})
+}
+func fileFormatConfigCSV(n string) string {
+	return fmt.Sprintf(`
+resource "snowflake_database" "test" {
+	name = "%v"
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_schema" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_file_format" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	schema = snowflake_schema.test.name
+	format_type = "CSV"
+	compression = "GZIP"
+	record_delimiter = "\r"
+	field_delimiter = ";"
+	file_extension = ".ssv"
+	skip_header = 1
+	skip_blank_lines = true
+	date_format = "YYY-MM-DD"
+	time_format = "HH24:MI"
+	timestamp_format = "YYYY-MM-DD HH24:MI:SS.FFTZH:TZM"
+	binary_format = "UTF8"
+	escape = "\\"
+	escape_unenclosed_field = "!"
+	trim_space = true
+	field_optionally_enclosed_by = "'" 
+	null_if = ["NULL"]
+	error_on_column_count_mismatch = true
+	replace_invalid_characters = true
+	validate_utf8 = false
+	empty_field_as_null = false 
+	skip_byte_order_mark = false
+	encoding = "UTF-16"
+	comment = "Terraform acceptance test"
+}
+`, n, n, n)
+}
+
+func fileFormatConfigJSON(n string) string {
+	return fmt.Sprintf(`
+resource "snowflake_database" "test" {
+	name = "%v"
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_schema" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_file_format" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	schema = snowflake_schema.test.name
+	format_type = "JSON"
+	compression = "GZIP"
+	date_format = "YYY-MM-DD"
+	time_format = "HH24:MI"
+	timestamp_format = "YYYY-MM-DD HH24:MI:SS.FFTZH:TZM"
+	binary_format = "UTF8"
+	trim_space = true
+	null_if = ["NULL"]
+	file_extension = ".jsn"
+	enable_octal = true
+	allow_duplicate = true
+	strip_outer_array = true
+	strip_null_values = true
+	ignore_utf8_errors = true
+	skip_byte_order_mark = false
+	comment = "Terraform acceptance test"
+}
+`, n, n, n)
+}
+
+func fileFormatConfigAvro(n string) string {
+	return fmt.Sprintf(`
+resource "snowflake_database" "test" {
+	name = "%v"
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_schema" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_file_format" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	schema = snowflake_schema.test.name
+	format_type = "AVRO"
+	compression = "GZIP"
+	trim_space = true
+	null_if = ["NULL"]
+	comment = "Terraform acceptance test"
+}
+`, n, n, n)
+}
+
+func fileFormatConfigORC(n string) string {
+	return fmt.Sprintf(`
+resource "snowflake_database" "test" {
+	name = "%v"
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_schema" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_file_format" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	schema = snowflake_schema.test.name
+	format_type = "ORC"
+	trim_space = true
+	null_if = ["NULL"]
+	comment = "Terraform acceptance test"
+}
+`, n, n, n)
+}
+
+func fileFormatConfigParquet(n string) string {
+	return fmt.Sprintf(`
+resource "snowflake_database" "test" {
+	name = "%v"
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_schema" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_file_format" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	schema = snowflake_schema.test.name
+	format_type = "PARQUET"
+	compression = "SNAPPY"
+	binary_as_text = true
+	trim_space = true
+	null_if = ["NULL"]
+	comment = "Terraform acceptance test"
+}
+`, n, n, n)
+}
+
+func fileFormatConfigXML(n string) string {
+	return fmt.Sprintf(`
+resource "snowflake_database" "test" {
+	name = "%v"
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_schema" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_file_format" "test" {
+	name = "%v"
+	database = snowflake_database.test.name
+	schema = snowflake_schema.test.name
+	format_type = "XML"
+	compression = "GZIP"
+	ignore_utf8_errors = true
+	preserve_space = true
+	strip_outer_element = true
+	disable_snowflake_data =  true
+	disable_auto_convert =  true
+	skip_byte_order_mark = false
+	comment = "Terraform acceptance test"
+}
+`, n, n, n)
+}
diff --git a/pkg/resources/file_format_test.go b/pkg/resources/file_format_test.go
new file mode 100644
index 00000000..14e3b624
--- /dev/null
+++ b/pkg/resources/file_format_test.go
@@ -0,0 +1,73 @@
+package resources_test
+
+import (
+	"database/sql"
+	"testing"
+
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/provider"
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/resources"
+	. "github.com/chanzuckerberg/terraform-provider-snowflake/pkg/testhelpers"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFileFormat(t *testing.T) {
+	r := require.New(t)
+	err := resources.FileFormat().InternalValidate(provider.Provider().Schema, true)
+	r.NoError(err)
+}
+
+func TestFileFormatCreate(t *testing.T) {
+	r := require.New(t)
+
+	in := map[string]interface{}{
+		"name":                           "test_file_format",
+		"database":                       "test_db",
+		"schema":                         "test_schema",
+		"format_type":                    "CSV",
+		"null_if":                        []interface{}{"NULL"},
+		"validate_utf8":                  true,
+		"error_on_column_count_mismatch": true,
+		"comment":                        "great comment",
+	}
+	d := schema.TestResourceDataRaw(t, resources.FileFormat().Schema, in)
+	r.NotNil(d)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		mock.ExpectExec(
+			`^CREATE FILE FORMAT "test_db"."test_schema"."test_file_format" TYPE = 'CSV' NULL_IF = \('NULL'\) SKIP_BLANK_LINES = false TRIM_SPACE = false ERROR_ON_COLUMN_COUNT_MISMATCH = true REPLACE_INVALID_CHARACTERS = false VALIDATE_UTF8 = true EMPTY_FIELD_AS_NULL = false SKIP_BYTE_ORDER_MARK = false COMMENT = 'great comment'$`,
+		).WillReturnResult(sqlmock.NewResult(1, 1))
+		expectReadFileFormat(mock)
+		err := resources.CreateFileFormat(d, db)
+		r.NoError(err)
+	})
+}
+
+func TestFileFormatCreateInvalidOptions(t *testing.T) {
+	r := require.New(t)
+
+	in := map[string]interface{}{
+		"name":          "test_file_format",
+		"database":      "test_db",
+		"schema":        "test_schema",
+		"format_type":   "JSON",
+		"null_if":       []interface{}{"NULL"},
+		"validate_utf8": true,
+		"comment":       "great comment",
+	}
+	d := schema.TestResourceDataRaw(t, resources.FileFormat().Schema, in)
+	r.NotNil(d)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		err := resources.CreateFileFormat(d, db)
+		r.EqualError(err, "validate_utf8 is an invalid format type option for format type JSON")
+	})
+}
+
+func expectReadFileFormat(mock sqlmock.Sqlmock) {
+	rows := sqlmock.NewRows([]string{
+		"created_on", "name", "database_name", "schema_name", "type", "owner", "comment", "format_options"},
+	).AddRow("2019-12-23 17:20:50.088 +0000", "test_file_format", "test_db", "test_schema", "CSV", "test", "great comment", `{"TYPE":"CSV","RECORD_DELIMITER":"\n","FIELD_DELIMITER":",","FILE_EXTENSION":null,"SKIP_HEADER":0,"DATE_FORMAT":"AUTO","TIME_FORMAT":"AUTO","TIMESTAMP_FORMAT":"AUTO","BINARY_FORMAT":"HEX","ESCAPE":"NONE","ESCAPE_UNENCLOSED_FIELD":"\\","TRIM_SPACE":false,"FIELD_OPTIONALLY_ENCLOSED_BY":"NONE","NULL_IF":["\\N"],"COMPRESSION":"AUTO","ERROR_ON_COLUMN_COUNT_MISMATCH":false,"VALIDATE_UTF8":false,"SKIP_BLANK_LINES":false,"REPLACE_INVALID_CHARACTERS":false,"EMPTY_FIELD_AS_NULL":false,"SKIP_BYTE_ORDER_MARK":false,"ENCODING":"UTF8"}`)
+	mock.ExpectQuery(`^SHOW FILE FORMATS LIKE 'test_file_format' IN DATABASE "test_db"$`).WillReturnRows(rows)
+}
diff --git a/pkg/snowflake/current_account_test.go b/pkg/snowflake/current_account_test.go
index 082e73eb..9b8c1717 100644
--- a/pkg/snowflake/current_account_test.go
+++ b/pkg/snowflake/current_account_test.go
@@ -1,11 +1,12 @@
 package snowflake_test
 
 import (
+	"testing"
+
 	sqlmock "github.com/DATA-DOG/go-sqlmock"
 	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/snowflake"
 	"github.com/jmoiron/sqlx"
 	"github.com/stretchr/testify/require"
-	"testing"
 )
 
 func TestCurrentAccountSelect(t *testing.T) {
diff --git a/pkg/snowflake/file_format_test.go b/pkg/snowflake/file_format_test.go
new file mode 100644
index 00000000..6bf3347c
--- /dev/null
+++ b/pkg/snowflake/file_format_test.go
@@ -0,0 +1,269 @@
+package snowflake
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestFileFormatCreateCSV(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(f.QualifiedName(), `"test_db"."test_schema"."test_file_format"`)
+
+	f.WithFormatType("CSV")
+	f.WithCompression("AUTO")
+	f.WithRecordDelimiter("\\n")
+	f.WithFieldDelimiter(",")
+	f.WithFileExtension(".CSV")
+	f.WithSkipHeader(1)
+	f.WithSkipBlankLines(true)
+	f.WithDateFormat("AUTO")
+	f.WithTimeFormat("AUTO")
+	f.WithTimestampFormat("AUTO")
+	f.WithBinaryFormat("HEX")
+	f.WithEscape("None")
+	f.WithEscapeUnenclosedField("None")
+	f.WithTrimSpace(true)
+	f.WithFieldOptionallyEnclosedBy("\"")
+	f.WithNullIf([]string{})
+	f.WithErrorOnColumnCountMismatch(false)
+	f.WithReplaceInvalidCharacters(false)
+	f.WithValidateUTF8(false)
+	f.WithEmptyFieldAsNull(true)
+	f.WithSkipByteOrderMark(true)
+	f.WithEncoding("UTF8")
+
+	r.Equal(`CREATE FILE FORMAT "test_db"."test_schema"."test_file_format" TYPE = 'CSV' COMPRESSION = 'AUTO' RECORD_DELIMITER = '\n' FIELD_DELIMITER = ',' FILE_EXTENSION = '.CSV' SKIP_HEADER = 1 DATE_FORMAT = 'AUTO' TIME_FORMAT = 'AUTO' TIMESTAMP_FORMAT = 'AUTO' BINARY_FORMAT = 'HEX' ESCAPE = 'None' ESCAPE_UNENCLOSED_FIELD = 'None' FIELD_OPTIONALLY_ENCLOSED_BY = '"' NULL_IF = () ENCODING = 'UTF8' SKIP_BLANK_LINES = true TRIM_SPACE = true ERROR_ON_COLUMN_COUNT_MISMATCH = false REPLACE_INVALID_CHARACTERS = false VALIDATE_UTF8 = false EMPTY_FIELD_AS_NULL = true SKIP_BYTE_ORDER_MARK = true`, f.Create())
+}
+
+func TestFileFormatCreateJSON(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format_json", "test_db", "test_schema")
+	r.Equal(f.QualifiedName(), `"test_db"."test_schema"."test_file_format_json"`)
+
+	f.WithFormatType("JSON")
+	f.WithCompression("AUTO")
+	f.WithDateFormat("AUTO")
+	f.WithTimeFormat("AUTO")
+	f.WithTimestampFormat("AUTO")
+	f.WithBinaryFormat("HEX")
+	f.WithTrimSpace(true)
+	f.WithNullIf([]string{"\\n", "NULL"})
+	f.WithAllowDuplicate(false)
+	f.WithStripOuterArray(false)
+	f.WithStripNullValues(false)
+	f.WithIgnoreUTF8Errors(true)
+
+	r.Equal(`CREATE FILE FORMAT "test_db"."test_schema"."test_file_format_json" TYPE = 'JSON' COMPRESSION = 'AUTO' DATE_FORMAT = 'AUTO' TIME_FORMAT = 'AUTO' TIMESTAMP_FORMAT = 'AUTO' BINARY_FORMAT = 'HEX' NULL_IF = ('\n', 'NULL') TRIM_SPACE = true ENABLE_OCTAL = false ALLOW_DUPLICATE = false STRIP_OUTER_ARRAY = false STRIP_NULL_VALUES = false REPLACE_INVALID_CHARACTERS = false IGNORE_UTF8_ERRORS = true SKIP_BYTE_ORDER_MARK = false`, f.Create())
+}
+
+func TestFileFormatChangeComment(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET COMMENT = 'worst format ever'`, f.ChangeComment("worst format ever"))
+}
+
+func TestFileFormatChangeCompression(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET COMPRESSION = 'GZIP'`, f.ChangeCompression("GZIP"))
+}
+
+func TestFileFormatChangeRecordDelimiter(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET RECORD_DELIMITER = '|'`, f.ChangeRecordDelimiter("|"))
+}
+
+func TestFileFormatChangeFieldDelimiter(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET FIELD_DELIMITER = '|'`, f.ChangeFieldDelimiter("|"))
+}
+
+func TestFileFormatChangeFileExtension(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET FILE_EXTENSION = '.csv.gz'`, f.ChangeFileExtension(".csv.gz"))
+}
+
+func TestFileFormatChangeDateFormat(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET DATE_FORMAT = 'AUTO'`, f.ChangeDateFormat("AUTO"))
+}
+
+func TestFileFormatChangeTimeFormat(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET TIME_FORMAT = 'AUTO'`, f.ChangeTimeFormat("AUTO"))
+}
+
+func TestFileFormatChangeTimestampFormat(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET TIMESTAMP_FORMAT = 'AUTO'`, f.ChangeTimestampFormat("AUTO"))
+}
+
+func TestFileFormatChangeBinaryFormat(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET BINARY_FORMAT = 'AUTO'`, f.ChangeBinaryFormat("AUTO"))
+}
+
+func TestFileFormatChangeValidateUTF8(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET VALIDATE_UTF8 = true`, f.ChangeValidateUTF8(true))
+}
+
+func TestFileFormatChangeEmptyFieldAsNull(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET EMPTY_FIELD_AS_NULL = true`, f.ChangeEmptyFieldAsNull(true))
+}
+
+func TestFileFormatChangeErrorOnColumnCountMismatch(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET ERROR_ON_COLUMN_COUNT_MISMATCH = true`, f.ChangeErrorOnColumnCountMismatch(true))
+}
+
+func TestFileFormatChangeEscape(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET ESCAPE = 'None'`, f.ChangeEscape("None"))
+}
+
+func TestFileFormatChangeFieldOptionallyEnclosedBy(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET FIELD_OPTIONALLY_ENCLOSED_BY = 'None'`, f.ChangeFieldOptionallyEnclosedBy("None"))
+}
+
+func TestFileFormatChangeEscapeUnenclosedField(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET ESCAPE_UNENCLOSED_FIELD = '\'`, f.ChangeEscapeUnenclosedField("\\"))
+}
+
+func TestFileFormatChangeNullIf(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET NULL_IF = ()`, f.ChangeNullIf([]string{}))
+}
+
+func TestFileFormatChangeEncoding(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET ENCODING = 'UTF8'`, f.ChangeEncoding("UTF8"))
+}
+
+func TestFileFormatChangeSkipHeader(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET SKIP_HEADER = 2`, f.ChangeSkipHeader(2))
+}
+
+func TestFileFormatChangeSkipBlankLines(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET SKIP_BLANK_LINES = true`, f.ChangeSkipBlankLines(true))
+}
+
+func TestFileFormatChangeTrimSpace(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET TRIM_SPACE = true`, f.ChangeTrimSpace(true))
+}
+
+func TestFileFormatChangeEnableOctal(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET ENABLE_OCTAL = true`, f.ChangeEnableOctal(true))
+}
+
+func TestFileFormatChangeAllowDuplicate(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET ALLOW_DUPLICATE = true`, f.ChangeAllowDuplicate(true))
+}
+
+func TestFileFormatChangeStripOuterArray(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET STRIP_OUTER_ARRAY = false`, f.ChangeStripOuterArray(false))
+}
+
+func TestFileFormatChangeStripNullValues(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET STRIP_NULL_VALUES = false`, f.ChangeStripNullValues(false))
+}
+
+func TestFileFormatChangeReplaceInvalidCharacters(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET REPLACE_INVALID_CHARACTERS = false`, f.ChangeReplaceInvalidCharacters(false))
+}
+
+func TestFileFormatChangeIgnoreUTF8Errors(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET IGNORE_UTF8_ERRORS = false`, f.ChangeIgnoreUTF8Errors(false))
+}
+
+func TestFileFormatChangeSkipByteOrderMark(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET SKIP_BYTE_ORDER_MARK = false`, f.ChangeSkipByteOrderMark(false))
+}
+
+func TestFileFormatChangeBinaryAsText(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET BINARY_AS_TEXT = false`, f.ChangeBinaryAsText(false))
+}
+
+func TestFileFormatChangePreserveSpace(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET PRESERVE_SPACE = false`, f.ChangePreserveSpace(false))
+}
+
+func TestFileFormatChangeStripOuterElement(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET STRIP_OUTER_ELEMENT = true`, f.ChangeStripOuterElement(true))
+}
+
+func TestFileFormatChangeDisableSnowflakeData(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET DISABLE_SNOWFLAKE_DATA = true`, f.ChangeDisableSnowflakeData(true))
+}
+
+func TestFileFormatChangeDisableAutoConvert(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`ALTER FILE FORMAT "test_db"."test_schema"."test_file_format" SET DISABLE_AUTO_CONVERT = true`, f.ChangeDisableAutoConvert(true))
+}
+
+func TestFileFormatDrop(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`DROP FILE FORMAT "test_db"."test_schema"."test_file_format"`, f.Drop())
+}
+
+func TestFileFormatDescribe(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`DESCRIBE FILE FORMAT "test_db"."test_schema"."test_file_format"`, f.Describe())
+}
+
+func TestFileFormatShow(t *testing.T) {
+	r := require.New(t)
+	f := FileFormat("test_file_format", "test_db", "test_schema")
+	r.Equal(`SHOW FILE FORMATS LIKE 'test_file_format' IN DATABASE "test_db"`, f.Show())
+}