Merge pull request #279 from TileDB-Inc/new-delete-fragments-endpoints

New delete fragments endpoints

Author: anastasop

.github/scripts/install_tiledb_linux.sh

@@ -1,4 +1,4 @@
 set -e -x
-curl --location -o tiledb.tar.gz https://github.com/TileDB-Inc/TileDB/releases/download/2.17.3/tiledb-linux-x86_64-2.17.3-0c2de58.tar.gz \
+curl --location -o tiledb.tar.gz https://github.com/TileDB-Inc/TileDB/releases/download/2.18.0/tiledb-linux-x86_64-2.18.0-71ca8b4.tar.gz \
 && sudo tar -C /usr/local -xf tiledb.tar.gz
 sudo ldconfig /usr/local/lib

.github/scripts/install_tiledb_linux_debug.sh

@@ -1,5 +1,5 @@
 set -e -x
-git clone https://github.com/TileDB-Inc/TileDB.git -b 2.17.3
+git clone https://github.com/TileDB-Inc/TileDB.git -b 2.18.0
 cd TileDB
 mkdir build && cd build
 cmake -DTILEDB_WERROR=OFF -DTILEDB_VCPKG=ON -DSANITIZER=leak -DTILEDB_VERBOSE=OFF -DTILEDB_S3=ON -DTILEDB_SERIALIZATION=ON -DCMAKE_BUILD_TYPE=Debug -DCMAKE_INSTALL_PREFIX=/usr/local ..

.github/scripts/install_tiledb_macos.sh

@@ -1,3 +1,3 @@
 set -e -x
-curl --location -o tiledb.tar.gz https://github.com/TileDB-Inc/TileDB/releases/download/2.17.3/tiledb-macos-x86_64-2.17.3-0c2de58.tar.gz \
+curl --location -o tiledb.tar.gz https://github.com/TileDB-Inc/TileDB/releases/download/2.18.0/tiledb-macos-x86_64-2.18.0-71ca8b4.tar.gz \
 && sudo tar -C /usr/local -xf tiledb.tar.gz

.github/scripts/install_tiledb_source_linux.sh

@@ -1,5 +1,5 @@
 set -e -x
-git clone https://github.com/TileDB-Inc/TileDB.git -b 2.17.3
+git clone https://github.com/TileDB-Inc/TileDB.git -b 2.18.0
 cd TileDB
 mkdir build && cd build
 cmake -DTILEDB_WERROR=OFF -DTILEDB_VCPKG=ON -DTILEDB_VERBOSE=OFF -DTILEDB_S3=ON -DTILEDB_SERIALIZATION=ON -DCMAKE_BUILD_TYPE=Release -DCMAKE_INSTALL_PREFIX=/usr/local ..

.github/scripts/install_tiledb_source_macos.sh

@@ -1,5 +1,5 @@
 set -e -x
-git clone https://github.com/TileDB-Inc/TileDB.git -b 2.17.3
+git clone https://github.com/TileDB-Inc/TileDB.git -b 2.18.0
 cd TileDB
 mkdir build && cd build
 cmake -DTILEDB_WERROR=OFF -DTILEDB_VCPKG=ON -DTILEDB_VERBOSE=OFF -DTILEDB_S3=ON -DTILEDB_SERIALIZATION=ON -DCMAKE_BUILD_TYPE=Release -DCMAKE_INSTALL_PREFIX=/usr/local ..

README.md

@@ -79,6 +79,7 @@ as such the below table reference which versions are compatible.
 | 0.21.0            | 2.15.X         |
 | 0.22.0            | 2.16.X         |
 | 0.23.0            | 2.17.X         |
+| 0.24.0            | 2.18.X         |
 
 
 ## Missing Functionality
@@ -102,3 +103,8 @@ removed in release 0.23. It is recommended to use the proper combination of
 The query methods `(Add|Get)?Range` are deprecated because they are deprecated in TileDB core.
 It is recommend to use the `Subarray` type for building queries.
 The methods will be removed in the release following their removal from TileDB core.
+
+### 0.24.0
+
+`Array.DeleteFragments` is deprecated in favor of `tiledb.DeleteFragments` which binds to
+`C.tiledb_array_delete_fragments_v2` the preferred method to delete fragments in TileDB 2.18.0.

array.go

@@ -1080,6 +1080,7 @@ func (a *Array) Config() (*Config, error) {
 }
 
 // DeleteFragments delete the range of fragments from startTimestamp to endTimestamp. The array needs to be opened modify exclusive.
+// Deprecated: Use the module DeleteFragments
 func (a *Array) DeleteFragments(startTimestamp, endTimestamp uint64) error {
 	curi := C.CString(a.uri)
 	defer C.free(unsafe.Pointer(curi))
@@ -1092,3 +1093,39 @@ func (a *Array) DeleteFragments(startTimestamp, endTimestamp uint64) error {
 
 	return nil
 }
+
+// DeleteFragments deletes the range of fragments from startTimestamp to endTimestamp
+func DeleteFragments(tdbCtx *Context, uri string, startTimestamp, endTimestamp uint64) error {
+	curi := C.CString(uri)
+	defer C.free(unsafe.Pointer(curi))
+
+	ret := C.tiledb_array_delete_fragments_v2(tdbCtx.tiledbContext, curi,
+		C.uint64_t(startTimestamp), C.uint64_t(endTimestamp))
+	if ret != C.TILEDB_OK {
+		return fmt.Errorf("Error deleting fragments from array: %s", tdbCtx.LastError())
+	}
+
+	return nil
+}
+
+// DeleteFragmentsList deletes the fragments of the list
+func DeleteFragmentsList(tdbCtx *Context, uri string, fragmentURIs []string) error {
+	curi := C.CString(uri)
+	defer C.free(unsafe.Pointer(curi))
+
+	var list []*C.char
+	for _, furi := range fragmentURIs {
+		cfuri := C.CString(furi)
+		defer C.free(unsafe.Pointer(cfuri))
+		list = append(list, cfuri)
+	}
+
+	ret := C.tiledb_array_delete_fragments_list(tdbCtx.tiledbContext, curi, (**C.char)(unsafe.Pointer(&list[0])), C.size_t(len(list)))
+	if ret != C.TILEDB_OK {
+		return fmt.Errorf("Error deleting fragments list from array: %s", tdbCtx.LastError())
+	}
+
+	runtime.KeepAlive(list)
+
+	return nil
+}

array_test.go

@@ -461,6 +461,127 @@ func TestArray_DeleteFragments(t *testing.T) {
 	require.Equal(t, int8(10), bounds[1])
 }
 
+func TestDeleteFragments(t *testing.T) {
+	// Create an array with domain [1, 10].
+	// Create fragments [1,2] [3,4] [5,6] [7,8] [9,10]
+	// Delete the first 2 fragments and verify the non empty domain is [5, 10]
+
+	// create an array and write 5 fragments
+	array, err := newTestArray(t)
+	require.NoError(t, err)
+
+	testStarted := time.Now()
+	var fragmentCreatedAt []time.Time
+
+	context, err := NewContext(nil)
+	require.NoError(t, err)
+	require.NoError(t, array.Open(TILEDB_WRITE))
+	for i := 1; i <= 10; i += 2 {
+		time.Sleep(100 * time.Millisecond) // give fragments some time distance
+
+		query, err := NewQuery(context, array)
+		require.NoError(t, err)
+		require.NotNil(t, query)
+
+		err = query.AddRangeByName("dim1", int32(i), int32(i+1))
+		require.NoError(t, err)
+		_, err = query.SetBuffer("a1", []int32{int32(i), int32(i + 1)})
+		_, _, err = query.SetBufferVar("a2", []uint64{0, 1}, []byte("aa"))
+		require.NoError(t, err)
+
+		err = query.Submit()
+		require.NoError(t, err)
+
+		status, err := query.Status()
+		require.NoError(t, err)
+		assert.Equal(t, TILEDB_COMPLETED, status)
+
+		fragmentCreatedAt = append(fragmentCreatedAt, time.Now())
+	}
+	err = array.Close()
+	require.NoError(t, err)
+
+	// delete the first two fragments
+	uri, err := array.URI()
+	require.NoError(t, err)
+	err = DeleteFragments(context, uri, uint64(testStarted.UnixMilli()), uint64(fragmentCreatedAt[1].UnixMilli()))
+	require.NoError(t, err)
+
+	// verify deletion
+	err = array.Open(TILEDB_READ)
+	require.NoError(t, err)
+	domain, _, err := array.NonEmptyDomainFromName("dim1")
+	require.NoError(t, err)
+	bounds := domain.Bounds.([]int8)
+	require.Equal(t, int8(5), bounds[0])
+	require.Equal(t, int8(10), bounds[1])
+}
+
+func TestDeleteFragmentsList(t *testing.T) {
+	// Create an array with domain [1, 10].
+	// Create fragments [1,2] [3,4] [5,6] [7,8] [9,10]
+	// Delete the first 2 fragments and verify the others
+
+	// create an array and write 5 fragments
+	array, err := newTestArray(t)
+	require.NoError(t, err)
+
+	context, err := NewContext(nil)
+	require.NoError(t, err)
+	err = array.Open(TILEDB_WRITE)
+	require.NoError(t, err)
+	for i := 1; i <= 10; i += 2 {
+		time.Sleep(100 * time.Millisecond)
+
+		query, err := NewQuery(context, array)
+		require.NoError(t, err)
+		require.NotNil(t, query)
+
+		err = query.AddRangeByName("dim1", int32(i), int32(i+1))
+		require.NoError(t, err)
+		_, err = query.SetBuffer("a1", []int32{int32(i), int32(i + 1)})
+		_, _, err = query.SetBufferVar("a2", []uint64{0, 1}, []byte("aa"))
+		require.NoError(t, err)
+
+		err = query.Submit()
+		require.NoError(t, err)
+
+		status, err := query.Status()
+		require.NoError(t, err)
+		assert.Equal(t, TILEDB_COMPLETED, status)
+	}
+	err = array.Close()
+	require.NoError(t, err)
+
+	uri, err := array.URI()
+	require.NoError(t, err)
+
+	getFragments := func() []string {
+		var fragmentURIs []string
+		fragmentInfo, err := NewFragmentInfo(context, uri)
+		require.NoError(t, err)
+		err = fragmentInfo.Load()
+		require.NoError(t, err)
+		fragmentNum, err := fragmentInfo.GetFragmentNum()
+		require.NoError(t, err)
+		for i := uint32(0); i < fragmentNum; i++ {
+			uri, err := fragmentInfo.GetFragmentURI(i)
+			require.NoError(t, err)
+			fragmentURIs = append(fragmentURIs, uri)
+		}
+		return fragmentURIs
+	}
+	fragmentURIsInitial := getFragments()
+
+	// delete the first two fragments
+	err = DeleteFragmentsList(context, uri, fragmentURIsInitial[0:2])
+	require.NoError(t, err)
+
+	// verify deletion
+	fragmentURIsAfter := getFragments()
+	require.ElementsMatch(t, fragmentURIsInitial[2:], fragmentURIsAfter)
+}
+
 func newTestArray(t *testing.T) (*Array, error) {
 	// Create configuration
 	config, err := NewConfig()

serialize.go

@@ -510,3 +510,27 @@ func DeserializeQueryAndArray(context *Context, buffer *Buffer, serializationTyp
 	runtime.KeepAlive(buffer)
 	return array, query, nil
 }
+
+// HandleArrayDeleteFragmentsTimestampsRequest is used by TileDB cloud to handle DeleteFragments with tiledb:// uris
+func HandleArrayDeleteFragmentsTimestampsRequest(context *Context, array *Array, buffer *Buffer, serializationType SerializationType) error {
+	ret := C.tiledb_handle_array_delete_fragments_timestamps_request(context.tiledbContext, array.tiledbArray,
+		C.tiledb_serialization_type_t(serializationType), buffer.tiledbBuffer)
+	if ret != C.TILEDB_OK {
+		return fmt.Errorf("error deserializing delete fragments timestamps: %s", context.LastError())
+	}
+
+	runtime.KeepAlive(buffer)
+	return nil
+}
+
+// HandleArrayDeleteFragmentsListRequest is used by TileDB cloud to handle DeleteFragmentsList with tiledb:// uris
+func HandleArrayDeleteFragmentsListRequest(context *Context, array *Array, buffer *Buffer, serializationType SerializationType) error {
+	ret := C.tiledb_handle_array_delete_fragments_list_request(context.tiledbContext, array.tiledbArray,
+		C.tiledb_serialization_type_t(serializationType), buffer.tiledbBuffer)
+	if ret != C.TILEDB_OK {
+		return fmt.Errorf("error deserializing delete fragments list: %s", context.LastError())
+	}
+
+	runtime.KeepAlive(buffer)
+	return nil
+}