init commit

Author: csimplestring

.DS_Store

@@ -0,0 +1,20 @@
+.vscode/
+
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Dependency directories (remove the comment below to include it)
+# vendor/
+
+# Go workspace file
+go.work

.gitignore

@@ -0,0 +1,86 @@
+# Delta Go
+
+## About
+
+This repository contains a Go package that provides a connector for [Delta Lake](https://delta.io/) - an open-source storage layer that brings ACID transactions to Apache Spark big data workloads. A Go portal of the official Scala [delta standalone](https://github.com/delta-io/connectors). 
+
+What is it?
+
+- It provides low level access to read and write Delta Lake **metadata** by implementing the Delta Lake transaction log protocol.
+
+What is it not?
+
+- It does not read, write or update the data for Delta Lake table directly, but the compute engine on top of it should do that.
+
+## Supported backends
+
+- Local file system (Done)
+- Azure Blob Storage (WIP)
+
+## Status
+
+- Currently only the local file system is fully supported and thoroughly tested against golden table data in the official Delta Standalone repo.
+- For other cloud storage, on the roadmap, any contribution is welcome!
+
+## Usage
+
+```go
+package examples
+
+import (
+	"log"
+	"path/filepath"
+	"testing"
+
+	delta "github.com/csimplestring/delta-go"
+)
+
+func main() {
+	path = "file://YOUR_DELTA_LOG_FOLDER/"
+
+	config := delta.Config{
+		StorageConfig: delta.StorageConfig{
+			Scheme: delta.Local,
+		},
+	}
+
+	table, err := delta.ForTable(path, config, &delta.SystemClock{})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// get the snapshot 
+	s, err := table.Snapshot()
+	if err != nil {
+		log.Fatal(err)
+	}
+	
+	// get the log version
+	version := s.Version()
+	log.Println(version)
+
+	// iterate all the log files
+	files, err := s.AllFiles()
+	if err != nil {
+		log.Fatal(err)
+	}
+	for _, f := range files {
+		log.Println(f.Path)
+	}
+}
+
+```
+
+## Contributing
+
+Contributions to this project are welcome. To contribute, please follow these steps:
+
+1. Fork the repository
+2. Create a new branch for your feature/bugfix
+3. Make changes and commit them with clear commit messages
+4. Push your changes to your forked repository
+5. Create a pull request to the original repository
+
+## License
+
+This project is licensed under the [Apache-2.0 License](https://www.apache.org/licenses/LICENSE-2.0).

README.md

@@ -0,0 +1,81 @@
+package action
+
+import (
+	"encoding/json"
+	"net/url"
+
+	"github.com/csimplestring/delta-go/errno"
+)
+
+const ReaderVersion = 1
+const WriterVersion = 2
+const MinReaderVersionProp = "delta.minReaderVersion"
+const MinWriterVersionProp = "delta.minWriterVersion"
+
+type Action interface {
+	Wrap() *SingleAction
+	Json() (string, error)
+}
+
+type FileAction interface {
+	Action
+	PathAsUri() (*url.URL, error)
+	IsDataChanged() bool
+}
+
+func FromJson(s string) (Action, error) {
+	action := &SingleAction{}
+	if err := json.Unmarshal([]byte(s), action); err != nil {
+		return nil, errno.JsonUnmarshalError(err)
+	}
+
+	return action.Unwrap(), nil
+}
+
+func jsonString(a Action) (string, error) {
+	b, err := json.Marshal(a.Wrap())
+	if err != nil {
+		return "", errno.JsonMarshalError(err)
+	}
+	return string(b), nil
+}
+
+func CheckMetadataProtocolProperties(metadata *Metadata, protocol *Protocol) error {
+	if _, ok := metadata.Configuration[MinReaderVersionProp]; ok {
+		return errno.AssertionError("should not have the protocol version MinReaderVersion as part of the table properties")
+	}
+	if _, ok := metadata.Configuration[MinWriterVersionProp]; ok {
+		return errno.AssertionError("should not have the protocol version MinWriterVersion as part of the table properties")
+	}
+	return nil
+}
+
+type SingleAction struct {
+	Txn        *SetTransaction `json:"txn,omitempty"`
+	Add        *AddFile        `json:"add,omitempty"`
+	Remove     *RemoveFile     `json:"remove,omitempty"`
+	MetaData   *Metadata       `json:"metaData,omitempty"`
+	Protocol   *Protocol       `json:"protocol,omitempty"`
+	Cdc        *AddCDCFile     `json:"cdc,omitempty"`
+	CommitInfo *CommitInfo     `json:"commitInfo,omitempty"`
+}
+
+func (s *SingleAction) Unwrap() Action {
+	if s.Add != nil {
+		return s.Add
+	} else if s.Remove != nil {
+		return s.Remove
+	} else if s.MetaData != nil {
+		return s.MetaData
+	} else if s.Txn != nil {
+		return s.Txn
+	} else if s.Protocol != nil {
+		return s.Protocol
+	} else if s.Cdc != nil {
+		return s.Cdc
+	} else if s.CommitInfo != nil {
+		return s.CommitInfo
+	} else {
+		return nil
+	}
+}

action/action.go

@@ -0,0 +1,46 @@
+package action
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMetadata_Equal(t *testing.T) {
+	t1 := int64(1)
+	t2 := int64(1)
+
+	m1 := &Metadata{
+		ID:          "1",
+		Name:        "2",
+		Description: "3",
+		Format: Format{
+			Proviver: "4",
+			Options: map[string]string{
+				"5": "5",
+			},
+		},
+		SchemaString:     "6",
+		PartitionColumns: []string{"7"},
+		Configuration:    map[string]string{"8": "8"},
+		CreatedTime:      &t1,
+	}
+
+	m2 := &Metadata{
+		ID:          "1",
+		Name:        "2",
+		Description: "3",
+		Format: Format{
+			Proviver: "4",
+			Options: map[string]string{
+				"5": "5",
+			},
+		},
+		SchemaString:     "6",
+		PartitionColumns: []string{"7"},
+		Configuration:    map[string]string{"8": "8"},
+		CreatedTime:      &t2,
+	}
+
+	assert.True(t, m1.Equals(m2))
+}

action/actions_test.go

@@ -0,0 +1,61 @@
+package action
+
+import (
+	"net/url"
+	"time"
+
+	"github.com/ulule/deepcopier"
+)
+
+type AddFile struct {
+	Path             string            `json:"path,omitempty"`
+	DataChange       bool              `json:"dataChange,omitempty"`
+	PartitionValues  map[string]string `json:"partitionValues,omitempty"`
+	Size             int64             `json:"size,omitempty"`
+	ModificationTime int64             `json:"modificationTime,omitempty"`
+	Stats            string            `json:"stats,omitempty"`
+	Tags             map[string]string `json:"tags,omitempty"`
+}
+
+func (a *AddFile) IsDataChanged() bool {
+	return a.DataChange
+}
+
+func (a *AddFile) PathAsUri() (*url.URL, error) {
+	return url.Parse(a.Path)
+}
+
+func (a *AddFile) Wrap() *SingleAction {
+	return &SingleAction{Add: a}
+}
+
+func (a *AddFile) Json() (string, error) {
+	return jsonString(a)
+}
+
+func (a *AddFile) Remove() *RemoveFile {
+	return a.RemoveWithTimestamp(nil, nil)
+}
+
+func (a *AddFile) RemoveWithTimestamp(ts *int64, dataChange *bool) *RemoveFile {
+	if ts == nil {
+		*ts = time.Now().UnixMilli()
+	}
+	if dataChange == nil {
+		*dataChange = true
+	}
+
+	return &RemoveFile{
+		Path:              a.Path,
+		DeletionTimestamp: ts,
+		DataChange:        *dataChange,
+	}
+}
+
+func (a *AddFile) Copy(dataChange bool, path string) *AddFile {
+	dst := &AddFile{}
+	deepcopier.Copy(a).To(dst)
+	dst.Path = path
+	dst.DataChange = dataChange
+	return dst
+}

action/add.go

@@ -0,0 +1,29 @@
+package action
+
+import (
+	"net/url"
+)
+
+type AddCDCFile struct {
+	Path            string            `json:"path,omitempty"`
+	DataChange      bool              `json:"dataChange,omitempty"`
+	PartitionValues map[string]string `json:"partitionValues,omitempty"`
+	Size            int64             `json:"size,omitempty"`
+	Tags            map[string]string `json:"tags,omitempty"`
+}
+
+func (a *AddCDCFile) IsDataChanged() bool {
+	return a.DataChange
+}
+
+func (a *AddCDCFile) PathAsUri() (*url.URL, error) {
+	return url.Parse(a.Path)
+}
+
+func (a *AddCDCFile) Wrap() *SingleAction {
+	return &SingleAction{Cdc: a}
+}
+
+func (a *AddCDCFile) Json() (string, error) {
+	return jsonString(a)
+}

action/cdc.go

@@ -0,0 +1,60 @@
+package action
+
+import (
+	"github.com/ulule/deepcopier"
+)
+
+type CommitMarker interface {
+	GetTimestamp() int64
+	WithTimestamp(timestamp int64) CommitMarker
+	GetVersion() int64
+}
+
+type CommitInfo struct {
+	Version             *int64            `json:"version,omitempty"`
+	Timestamp           int64             `json:"timestamp,omitempty"`
+	UserID              *string           `json:"userId,omitempty"`
+	UserName            *string           `json:"userName,omitempty"`
+	Operation           string            `json:"operation,omitempty"`
+	OperationParameters map[string]string `json:"operationParameters,omitempty"`
+	Job                 *JobInfo          `json:"job,omitempty"`
+	Notebook            *NotebookInfo     `json:"notebook,omitempty"`
+	ClusterId           *string           `json:"clusterId,omitempty"`
+	ReadVersion         *int64            `json:"readVersion,omitempty"`
+	IsolationLevel      *string           `json:"isolationLevel,omitempty"`
+	IsBlindAppend       *bool             `json:"isBlindAppend,omitempty"`
+	OperationMetrics    map[string]string `json:"operationMetrics,omitempty"`
+	UserMetadata        *string           `json:"userMetadata,omitempty"`
+	EngineInfo          *string           `json:"engineInfo,omitempty"`
+}
+
+func (c *CommitInfo) Wrap() *SingleAction {
+	return &SingleAction{CommitInfo: c}
+}
+
+func (c *CommitInfo) Json() (string, error) {
+	return jsonString(c)
+}
+
+func (c *CommitInfo) GetTimestamp() int64 {
+	return c.Timestamp
+}
+
+func (c *CommitInfo) WithTimestamp(timestamp int64) CommitMarker {
+	copied := &CommitInfo{}
+	deepcopier.Copy(c).To(copied)
+
+	copied.Timestamp = timestamp
+	return copied
+}
+
+func (c *CommitInfo) GetVersion() int64 {
+	return *c.Version
+}
+
+func (c *CommitInfo) Copy(version int64) *CommitInfo {
+	res := &CommitInfo{}
+	deepcopier.Copy(c).To(res)
+	res.Version = &version
+	return res
+}

action/commit_info.go

@@ -0,0 +1,6 @@
+package action
+
+type Format struct {
+	Proviver string            `json:"provider,omitempty"`
+	Options  map[string]string `json:"options,omitempty"`
+}