rymurr commented on a change in pull request #1216:
URL: https://github.com/apache/iceberg/pull/1216#discussion_r475213365
##########
File path: python/iceberg/hive/hive_table_operations.py
##########
@@ -50,10 +63,146 @@ def refresh(self):
return self.current()
def commit(self, base, metadata):
- raise NotImplementedError()
+ new_metadata_location = self.write_new_metadata(metadata, self.version
+ 1)
+
+ threw = True
+ lock_id = None
+ try:
+ lock_id = self.acquire_lock()
+ if base is not None:
+ with self._client as open_client:
+ tbl = open_client.get_table(self.database, self.table)
+ else:
+ current_time_millis = int(time.time())
+ tbl = Table(self.table,
+ self.database,
+ getpass.getuser(),
+ current_time_millis // 1000,
+ current_time_millis // 1000,
+ sd=storage_descriptor(metadata),
+ tableType="EXTERNAL_TABLE",
+ parameters={'EXTERNAL': 'TRUE'})
+
+ tbl.sd = storage_descriptor(metadata)
+ metadata_location =
tbl.parameters.get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP, None)
+ base_metadata_location = base.metadataFileLocation() if base is
not None else None
+ if base_metadata_location != metadata_location:
+ raise CommitFailedException(
+ "Base metadata location '%s' is not same as the current
table metadata location '%s' for %s.%s",
+ base_metadata_location, metadata_location, self.database,
self.table)
+
+ self.set_parameters(new_metadata_location, tbl)
+
+ if base is not None:
+ with self._client as open_client:
+ env_context = EnvironmentContext(
+ {"DO_NOT_UPDATE_STATS": "true"}
+ )
+ open_client.alter_table(self.database, self.table, tbl,
env_context)
+
+ else:
+ with self._client as open_client:
+ open_client.create_table(tbl)
+ threw = False
+ except AlreadyExistsException:
+ raise IcebergAlreadyExistsException("Table already exists:
{}.{}".format(self.database, self.table))
+ except TException as e:
+ if e is not None and "Table/View 'HIVE_LOCKS' does not exist" in
str(e):
+ raise Exception("""Failed to acquire locks from metastore
because 'HIVE_LOCKS' doesn't
+ exist, this probably happened when using
embedded metastore or doesn't create a
+ transactional meta table. To fix this, use an
alternative metastore""", e)
+
+ raise Exception("Metastore operation failed for
{}.{}".format(self.database, self.table), e)
+ finally:
+ if threw:
+ self.io().delete(new_metadata_location)
+ self.unlock(lock_id)
+
+ def set_parameters(self, new_metadata_location, tbl):
+ parameters = tbl.parameters
+
+ if not parameters:
+ parameters = dict()
+
+ parameters[BaseMetastoreTableOperations.TABLE_TYPE_PROP] = \
+ BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE.upper()
+ parameters[BaseMetastoreTableOperations.METADATA_LOCATION_PROP] =
new_metadata_location
+
+ if self.current_metadata_location is not None and
len(self.current_metadata_location) > 0:
+
parameters[BaseMetastoreTableOperations.PREVIOUS_METADATA_LOCATION_PROP] =
self.current_metadata_location
+
+ tbl.parameters = parameters
+
+ def unlock(self, lock_id):
+ if lock_id:
+ try:
+ self.do_unlock(LockResponse(lock_id))
+ except Exception as e:
+ logging.warning("Failed to unlock {}.{}".format(self.database,
self.table), e)
+
+ def do_unlock(self, lock_id):
Review comment:
yup, fixed
##########
File path: python/iceberg/hive/hive_table_operations.py
##########
@@ -50,10 +63,146 @@ def refresh(self):
return self.current()
def commit(self, base, metadata):
- raise NotImplementedError()
+ new_metadata_location = self.write_new_metadata(metadata, self.version
+ 1)
+
+ threw = True
+ lock_id = None
+ try:
+ lock_id = self.acquire_lock()
+ if base is not None:
+ with self._client as open_client:
+ tbl = open_client.get_table(self.database, self.table)
+ else:
+ current_time_millis = int(time.time())
+ tbl = Table(self.table,
+ self.database,
+ getpass.getuser(),
+ current_time_millis // 1000,
+ current_time_millis // 1000,
+ sd=storage_descriptor(metadata),
+ tableType="EXTERNAL_TABLE",
+ parameters={'EXTERNAL': 'TRUE'})
+
+ tbl.sd = storage_descriptor(metadata)
+ metadata_location =
tbl.parameters.get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP, None)
+ base_metadata_location = base.metadataFileLocation() if base is
not None else None
+ if base_metadata_location != metadata_location:
+ raise CommitFailedException(
+ "Base metadata location '%s' is not same as the current
table metadata location '%s' for %s.%s",
+ base_metadata_location, metadata_location, self.database,
self.table)
+
+ self.set_parameters(new_metadata_location, tbl)
+
+ if base is not None:
+ with self._client as open_client:
+ env_context = EnvironmentContext(
+ {"DO_NOT_UPDATE_STATS": "true"}
+ )
+ open_client.alter_table(self.database, self.table, tbl,
env_context)
+
+ else:
+ with self._client as open_client:
+ open_client.create_table(tbl)
+ threw = False
+ except AlreadyExistsException:
+ raise IcebergAlreadyExistsException("Table already exists:
{}.{}".format(self.database, self.table))
+ except TException as e:
+ if e is not None and "Table/View 'HIVE_LOCKS' does not exist" in
str(e):
+ raise Exception("""Failed to acquire locks from metastore
because 'HIVE_LOCKS' doesn't
+ exist, this probably happened when using
embedded metastore or doesn't create a
+ transactional meta table. To fix this, use an
alternative metastore""", e)
+
+ raise Exception("Metastore operation failed for
{}.{}".format(self.database, self.table), e)
+ finally:
+ if threw:
+ self.io().delete(new_metadata_location)
+ self.unlock(lock_id)
+
+ def set_parameters(self, new_metadata_location, tbl):
+ parameters = tbl.parameters
+
+ if not parameters:
+ parameters = dict()
+
+ parameters[BaseMetastoreTableOperations.TABLE_TYPE_PROP] = \
+ BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE.upper()
+ parameters[BaseMetastoreTableOperations.METADATA_LOCATION_PROP] =
new_metadata_location
+
+ if self.current_metadata_location is not None and
len(self.current_metadata_location) > 0:
+
parameters[BaseMetastoreTableOperations.PREVIOUS_METADATA_LOCATION_PROP] =
self.current_metadata_location
+
+ tbl.parameters = parameters
+
+ def unlock(self, lock_id):
+ if lock_id:
+ try:
+ self.do_unlock(LockResponse(lock_id))
+ except Exception as e:
+ logging.warning("Failed to unlock {}.{}".format(self.database,
self.table), e)
+
+ def do_unlock(self, lock_id):
+ with self._client as open_client:
+ open_client.unlock(lock_id)
+
+ def acquire_lock(self):
+ lock_component = LockComponent(LockType.EXCLUSIVE, LockLevel.TABLE,
self.database, self.table)
+
+ lock_request = LockRequest([lock_component], user=getpass.getuser(),
hostname=socket.gethostname())
+ with self._client as open_client:
+ lock_response = open_client.lock(lock_request)
+
+ state = lock_response.state
+ lock_id = lock_response.lockid
+ start = int(time.time())
+ duration = 0
+ timeout = False
Review comment:
agreed, fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
