rdblue commented on a change in pull request #1216:
URL: https://github.com/apache/iceberg/pull/1216#discussion_r467343967
##########
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
+ while not timeout and state == LockState.WAITING:
+ with self._client as open_client:
+ lock_response = open_client.check_lock(lock_response)
+ state = lock_response.state
+
+ duration = int(time.time()) - start
+ if duration > 3 * 60 * 1000:
+ timeout = True
+ else:
+ time.sleep(0.05)
+
+ if timeout and state != LockState.ACQUIRED:
+ raise CommitFailedException("Timed out after {} ms waiting for
lock on {}.{}".format(duration,
+
self.database,
+
self.table))
+
+ if state != LockState.ACQUIRED:
+ raise CommitFailedException(
+ "Could not acquire the lock on {}.{}, lock request ended in
state {}".format(self.database, self.table,
+
state))
+ return lock_id
def io(self):
- raise NotImplementedError()
+ return get_fs(self.base_location, self.conf)
def close(self):
self._client.close()
+
+
+def storage_descriptor(metadata):
+ ser_de_info =
SerDeInfo(serializationLib="org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe")
+ return StorageDescriptor(columns(metadata.schema),
+ metadata.location,
+ "org.apache.hadoop.mapred.FileInputFormat",
+ "org.apache.hadoop.mapred.FileOutputFormat",
+ serdeInfo=ser_de_info)
+
+
+def columns(schema):
+ return [FieldSchema(col.name, convert_hive_type(col.type), "") for col in
schema.columns()]
+
+
+def convert_hive_type(col_type):
+ try:
+ type_id = col_type.type_id.value['hive_name']
+ if type_id is not None:
+ return type_id
+ except: # NOQA
+ raise NotImplementedError("Not yet implemented column type " +
col_type)
Review comment:
Since this is identical to the fall through case, can this be `pass`
instead?
----------------------------------------------------------------
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]
